"""Trading Agent — official Python client for the Quant API (v1). Zero dependencies (uses the standard library). Optional: pandas, for the ``*_df`` helpers that return DataFrames. Quickstart ---------- pip install pandas # optional, only for the _df helpers export TRADING_AGENT_API_KEY=ta_live_... from tradingagent import TradingAgent ta = TradingAgent() # reads the env var # 1. Pull the resolved signal panel (point-in-time, factual history). signals = ta.signals(market="TW", horizon="21d", limit=5000) print(signals["count"], "rows") # 2. Backtest a long/short strategy on our archive, after real costs. bt = ta.backtest( config={"direction": "long_short", "quantile": 0.1, "weighting": "equal", "holdDays": 21}, source={"market": "TW", "horizon": "21d"}, ) m = bt["metrics"] print(f"Sharpe {m['sharpe']:.2f} CAGR {m['cagr']:.1%} maxDD {m['max_drawdown']:.1%}") # 3. Sweep parameters and get the overfit-corrected Deflated Sharpe. sw = ta.backtest(config={"holdDays": 21}, source={"market": "US", "horizon": "21d"}, sweep=True) print("best:", sw["best"]["label"], "DSR:", sw["deflated_sharpe"]["dsr"]) Create an API key at https://www.tradingagentapp.com/account/api-keys (the signal panel needs any key; the backtest endpoint needs a Quant key). """ from __future__ import annotations import json import os import urllib.error import urllib.parse import urllib.request from typing import Any, Dict, List, Optional __version__ = "1.0.0" DEFAULT_BASE_URL = "https://www.tradingagentapp.com" class TradingAgentError(Exception): """Raised on auth failures, HTTP errors, or bad responses.""" def __init__(self, message: str, status: Optional[int] = None, body: Any = None): super().__init__(message) self.status = status self.body = body class TradingAgent: """Client for the Trading Agent Quant API. Parameters ---------- api_key: Your ``ta_live_…`` key. Falls back to the ``TRADING_AGENT_API_KEY`` environment variable. base_url: Override the API host (defaults to production). timeout: Per-request timeout in seconds. """ def __init__( self, api_key: Optional[str] = None, base_url: str = DEFAULT_BASE_URL, timeout: int = 60, ): self.api_key = api_key or os.environ.get("TRADING_AGENT_API_KEY") if not self.api_key: raise TradingAgentError( "API key required. Pass api_key=... or set TRADING_AGENT_API_KEY." ) self.base_url = base_url.rstrip("/") self.timeout = timeout # -- internals ----------------------------------------------------------- def _request( self, method: str, path: str, params: Optional[Dict[str, Any]] = None, body: Optional[Dict[str, Any]] = None, ) -> Any: url = self.base_url + path if params: clean = {k: v for k, v in params.items() if v is not None} if clean: url += "?" + urllib.parse.urlencode(clean) data = json.dumps(body).encode("utf-8") if body is not None else None req = urllib.request.Request(url, data=data, method=method) req.add_header("Authorization", f"Bearer {self.api_key}") req.add_header("Accept", "application/json") req.add_header("User-Agent", f"tradingagent-python/{__version__}") if data is not None: req.add_header("Content-Type", "application/json") try: with urllib.request.urlopen(req, timeout=self.timeout) as resp: return json.loads(resp.read().decode("utf-8")) except urllib.error.HTTPError as e: try: detail = json.loads(e.read().decode("utf-8")) except Exception: detail = {"message": e.reason} raise TradingAgentError( detail.get("message") or detail.get("error") or f"HTTP {e.code}", status=e.code, body=detail, ) from None except urllib.error.URLError as e: raise TradingAgentError(f"Network error: {e.reason}") from None # -- signals ------------------------------------------------------------- def signals( self, market: Optional[str] = None, horizon: Optional[str] = None, date_from: Optional[str] = None, date_to: Optional[str] = None, limit: int = 5000, ) -> Dict[str, Any]: """Fetch resolved signal rows. Returns the parsed JSON envelope ``{count, data, schema, tier, full_access, delayed_days}``.""" return self._request( "GET", "/api/v1/signals", params={"market": market, "horizon": horizon, "from": date_from, "to": date_to, "limit": limit}, ) def signals_df(self, **kwargs: Any): """Like :meth:`signals` but returns a pandas DataFrame of the rows.""" import pandas as pd # lazy import — only needed for this helper return pd.DataFrame(self.signals(**kwargs)["data"]) # -- backtest ------------------------------------------------------------ def backtest( self, config: Optional[Dict[str, Any]] = None, source: Optional[Dict[str, Any]] = None, panel: Optional[List[Dict[str, Any]]] = None, sweep: bool = False, grid: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: """Run a cross-sectional strategy backtest (Quant tier). Provide ``source`` to backtest our resolved archive, or ``panel`` to backtest your own rows (each: ``{date, asset, market?, score, forward_return}``). Set ``sweep=True`` for a parameter sweep with the overfit-corrected Deflated Sharpe. """ body: Dict[str, Any] = {"config": config or {}, "sweep": sweep} if panel is not None: body["panel"] = panel if source is not None: body["source"] = source if grid is not None: body["grid"] = grid return self._request("POST", "/api/v1/backtest", body=body) # -- fundamentals -------------------------------------------------------- def fundamentals( self, ticker: Optional[str] = None, as_of: Optional[str] = None, metric: Optional[str] = None, date_from: Optional[str] = None, view: Optional[str] = None, ) -> Dict[str, Any]: """Point-in-time company fundamentals from SEC EDGAR (as-first-reported). Omit ``ticker`` for the index (available tickers + metrics). Pass ``as_of`` (ISO date) to get only filings public by that date — a backtest at date T should use ``as_of=T``. ``view="latest"`` returns the as-of snapshot (latest value per metric) you'd join onto a price/signal. """ return self._request( "GET", "/api/v1/fundamentals", params={"ticker": ticker, "as_of": as_of, "metric": metric, "from": date_from, "view": view}, ) def fundamentals_df(self, ticker: str, **kwargs: Any): """Like :meth:`fundamentals` but returns a pandas DataFrame of the rows.""" import pandas as pd return pd.DataFrame(self.fundamentals(ticker=ticker, **kwargs).get("data", [])) __all__ = ["TradingAgent", "TradingAgentError", "__version__"]