Programmatic usage

Use TradingGraph directly in Python to run QuantAgent’s multi-agent analysis pipeline without the web interface.

TradingGraph is the main entry point for programmatic use. It initialises the LLMs, builds the LangGraph workflow, and exposes a single .graph.invoke() call that accepts your OHLCV data and returns a complete analysis state.

Installation check

Confirm the package is importable before writing integration code:

from trading_graph import TradingGraph

If this raises an ImportError, ensure you have activated the project’s conda environment and installed all dependencies (pip install -r requirements.txt).

Initialising TradingGraph

Default config (OpenAI)

from trading_graph import TradingGraph

# Uses DEFAULT_CONFIG: OpenAI provider, gpt-4o-mini agent, gpt-4o graph model
trading_graph = TradingGraph()

When no config is provided, TradingGraph falls back to DEFAULT_CONFIG from default_config.py and reads the API key from the OPENAI_API_KEY environment variable.

Building the initial state

The graph expects a dictionary with the following keys:

Key	Type	Description
kline_data	dict	OHLCV data with keys Datetime, Open, High, Low, Close — each a list of scalar values.
analysis_results	None	Set to None on first invocation; the graph populates this field.
messages	list	Empty list []; used internally by the agent message passing layer.
time_frame	str	Human-readable timeframe label, e.g. "4hour", "1min", "1 week".
stock_name	str	Display name of the asset, e.g. "BTC" or "S&P 500".

Preparing kline_data from a DataFrame

The web interface uses pandas DataFrames fetched from Yahoo Finance. Convert your own DataFrame to the required dict format like this:

import pandas as pd

# df must have columns: Datetime, Open, High, Low, Close
required_columns = ["Datetime", "Open", "High", "Low", "Close"]
df_slice = df.tail(45).reset_index(drop=True)

kline_data = {}
for col in required_columns:
    if col == "Datetime":
        kline_data[col] = df_slice[col].dt.strftime("%Y-%m-%d %H:%M:%S").tolist()
    else:
        kline_data[col] = df_slice[col].tolist()

The pipeline analyses the most recent 45 candles. Passing significantly fewer candles may reduce indicator accuracy; passing more is unnecessary as the web interface already slices to 45.

Running analysis and reading results

from trading_graph import TradingGraph
import yfinance as yf
import pandas as pd

# --- 1. Fetch data ---
df = yf.download("BTC-USD", start="2025-03-01", end="2025-03-28", interval="4h")
df = df.reset_index()
if hasattr(df.columns, "get_level_values"):
    df.columns = df.columns.get_level_values(0)
df = df.rename(columns={"Date": "Datetime"})
df["Datetime"] = pd.to_datetime(df["Datetime"])

# --- 2. Prepare kline_data ---
required_columns = ["Datetime", "Open", "High", "Low", "Close"]
df_slice = df[required_columns].tail(45).reset_index(drop=True)

kline_data = {}
for col in required_columns:
    if col == "Datetime":
        kline_data[col] = df_slice[col].dt.strftime("%Y-%m-%d %H:%M:%S").tolist()
    else:
        kline_data[col] = df_slice[col].tolist()

# --- 3. Build initial state ---
initial_state = {
    "kline_data": kline_data,
    "analysis_results": None,
    "messages": [],
    "time_frame": "4hour",
    "stock_name": "BTC",
}

# --- 4. Run the graph ---
trading_graph = TradingGraph()
final_state = trading_graph.graph.invoke(initial_state)

# --- 5. Access results ---
print(final_state.get("final_trade_decision"))  # JSON string: LONG/SHORT + metadata
print(final_state.get("indicator_report"))       # Technical indicator analysis
print(final_state.get("pattern_report"))         # Chart pattern detection
print(final_state.get("trend_report"))           # Trend channel analysis

Result fields

Field	Type	Contents
final_trade_decision	str	JSON string with decision (LONG/SHORT), risk_reward_ratio, forecast_horizon, and justification.
indicator_report	str	Free-text summary of RSI, MACD, Stochastic, and other indicator values.
pattern_report	str	Plain-language description of the detected chart pattern.
trend_report	str	Trend direction, channel slope, ADX reading, and consolidation assessment.

Parse the decision field as JSON:

import json

raw = final_state.get("final_trade_decision", "{}")
start = raw.find("{")
end = raw.rfind("}") + 1
if start != -1 and end:
    decision = json.loads(raw[start:end])
    print(decision["decision"])          # "LONG" or "SHORT"
    print(decision["risk_reward_ratio"]) # e.g. "1:2.5"
    print(decision["forecast_horizon"])  # e.g. "24-48 hours"
    print(decision["justification"])

Updating API keys at runtime

You can rotate or set an API key on an existing TradingGraph instance without recreating it. This is equivalent to what the web interface does when you save a key in the Settings panel:

# Switch to a new OpenAI key
trading_graph.update_api_key("sk-new-key-here", provider="openai")

# Switch to an Anthropic key
trading_graph.update_api_key("sk-ant-...", provider="anthropic")

# Switch to a Qwen key
trading_graph.update_api_key("sk-...", provider="qwen")

update_api_key() writes the key to both the config dict and the corresponding environment variable (OPENAI_API_KEY, ANTHROPIC_API_KEY, or DASHSCOPE_API_KEY), then calls refresh_llms() to rebuild the LLM objects and recompile the graph.