TradingGraph - Medium-to-low frequency stock strategy workflows

Coding with DeepSeek

When I used AI to develop trading strategies, the generated code quickly became complex and hard to manage. What if I could simplify it using a workflow-based approach — composing small, typed nodes into a clear, executable pipeline? So instead of letting AI generate another tangled codebase, I had it help me build tradinggraph: a thin orchestration layer that turns strategy logic into composable, compile-time-validated workflows.

In Development

• Class nodes — wrap stateful strategies with constructor-injected config
• Conditional routing — dynamic branch selection based on upstream node output
• Actor-based execution — each node runs as an isolated actor
• JSON serialization — export and reconstruct graphs via to_json() / from_json()

See It in Action

A workflow covers a single strategy chain: stock screening → market environment assessment → entry timing → position sizing → risk control → order submission.

flowchart LR
    A["stock_list"] --> B["fundamental"]
    B --> C["technical"]
    D["market_data"] --> E["market_env"]
    C --> F["entry"]
    E --> F
    F --> G{"signal"}
    G -->|BUY| H["sizing"]
    H --> J{"risk"}
    J -->|APPROVED| L["order"]
    J -->|REJECTED| M((end))
    G -->|HOLD| N((end))

Loading

Quick Start

from tradinggraph import Graph, ActorGraphExecutor, Context, node

@node
async def fetch_sh(ctx: Context) -> StockList:
    return ["600000", "600519"]

@node
async def fetch_sz(ctx: Context) -> StockList:
    return ["000001", "000858"]

@node
async def merge_stocks(
    ctx: Context,
    sh_stocks: StockList,
    sz_stocks: StockList,
) -> StockList:
    return sh_stocks + sz_stocks

g = Graph("spot_merger")
g.add_node("fetch_sh", fetch_sh)
g.add_node("fetch_sz", fetch_sz)
g.add_node("merge", merge_stocks)

g.add_edge("fetch_sh", "merge", param="sh_stocks")
g.add_edge("fetch_sz", "merge", param="sz_stocks")