""" Pre-compute backtest results for all strategies, for multiple capital scenarios. Each strategy is backtested ONCE per market with that market's live starting capital (MONITOR_CAPITAL: US $10k, CN ¥70k); the resulting equity series is then replayed with different cash-injection schedules so we get multiple scenarios without re-running each strategy. Usage: uv run python precompute_backtest.py [--output PATH] [--years N] Output JSON shape: { "computedAt": "...", "scenarios": { "lump_sum_10k": {label, capital, injectionSchedule, markets: {us, cn}}, "dca_feb_aug": {label, capital, injectionSchedule, markets: {us, cn}} } } """ import argparse import json import os import sys from datetime import datetime import pandas as pd from backtest_runner import run_market, ENGINE_VERSION from bridge import ( STRATEGY_META, _build_strategy_result, _sanitize_for_json, ) from trader import AGENT_REVIEW_STRATEGIES, MONITOR_CAPITAL, STRATEGY_REGISTRY, strategy_supports_market # Initial capital per market, in local currency — the same amounts the live # monitor starts with (US $10k, CN ¥70k ≈ $10k). Scenario amounts below are # multiples of it, so both markets model the same real money. MARKET_CAPITAL = dict(MONITOR_CAPITAL) SCENARIOS = [ { "id": "lump_sum_10k", "label": "起始资金一次性投入", "description": "US $10,000 / CN ¥70,000 一次性投入，无后续追加", "injection": None, }, { "id": "dca_feb_aug", "label": "起始资金 + 每年 2/8 月追加一半", "description": "US $10,000 + 每年 2/8 月各 $5,000；CN ¥70,000 + 每年 2/8 月各 ¥35,000", "injection": { "amount_ratio": 0.5, "frequency": "yearly_months", "months": [2, 8], "startYear": 1, }, }, ] def _injection_for_market(injection: dict | None, capital: float) -> dict | None: """Materialize a scenario's injection schedule in a market's currency.""" if not injection: return None out = {k: v for k, v in injection.items() if k != "amount_ratio"} out["amount"] = round(capital * injection["amount_ratio"], 2) return out def log(msg: str) -> None: print(f"[precompute] {msg}", file=sys.stderr) def _equity_series_from_curve(curve: list[dict]) -> pd.Series: """Reconstruct a pandas Series from a precomputed equity curve list.""" if not curve: return pd.Series(dtype=float) dates = pd.to_datetime([p["date"] for p in curve]) values = [float(p["value"]) for p in curve] return pd.Series(values, index=dates) def build_scenario_for_market( raw: dict, scenario: dict, years: int, capital: float, ) -> dict: """Materialize a scenario (with optional injections) for one market.""" injection_schedule = _injection_for_market(scenario.get("injection"), capital) benchmark_curve = raw["benchmark_curve"] benchmark_label = raw["benchmark_label"] strategy_results = [] for run in raw["strategy_runs"]: equity = run["equity"] try: result = _build_strategy_result( run["name"], equity, run["trades"], injection_schedule, capital, execution_plan=run.get("_plan"), fee_params=run.get("_fees"), ) strategy_results.append(result) except Exception as e: log(f" [scenario={scenario['id']}] FAILED {run['name']}: {e}") # Re-scale benchmark for DCA scenarios so the comparison stays apples-to-apples. final_bench_curve = benchmark_curve if injection_schedule and benchmark_curve: bench_series = _equity_series_from_curve(benchmark_curve) bench_returns = bench_series.pct_change().fillna(0.0) from bridge import build_injection_schedule from research.permanent_yearly import equity_with_cashflows contribs, _ = build_injection_schedule(bench_series, injection_schedule) bench_with_inj = equity_with_cashflows(bench_returns, contribs, capital) final_bench_curve = [ {"date": d.strftime("%Y-%m-%d"), "value": round(float(v), 2)} for d, v in bench_with_inj.items() ] return { "meta": { "market": raw["market"], "years": raw.get("actual_years", years), "requestedYears": years, "initialCapital": capital, "startDate": raw["start_date"], "endDate": raw["end_date"], "txCost": raw["tx_cost"], "fixedFee": raw["fixed_fee"], "injectionSchedule": injection_schedule, "engineVersion": raw.get("engine_version"), "warnings": raw.get("warnings", []), "coverage": raw.get("coverage"), "errors": raw.get("errors", []), }, "strategies": strategy_results, "benchmark": { "name": benchmark_label, "equityCurve": final_bench_curve, }, } def main(): parser = argparse.ArgumentParser(description="Pre-compute backtest results") parser.add_argument( "--output", default=None, help="Output directory (default: data/precomputed, read by the web UI)", ) parser.add_argument( "--years", type=int, default=10, help="Backtest period in years (default: 10)", ) parser.add_argument("--offline", action="store_true", help="Use the frozen local data snapshot") args = parser.parse_args() output_dir = args.output or os.path.join( os.path.dirname(os.path.abspath(__file__)), "data", "precomputed" ) os.makedirs(output_dir, exist_ok=True) us_strategies = [] cn_strategies = [] for name in STRATEGY_REGISTRY: if name in AGENT_REVIEW_STRATEGIES: # forward only: no history before its first score file continue meta = STRATEGY_META.get(name, {"markets": ["us", "cn"]}) markets = [m for m in meta.get("markets", ["us", "cn"]) if strategy_supports_market(name, m)] if "us" in markets: us_strategies.append(name) if "cn" in markets: cn_strategies.append(name) us_capital = MARKET_CAPITAL["us"] cn_capital = MARKET_CAPITAL["cn"] # Run each market once (cached price data → fast subsequent runs across scenarios). log("=" * 60) log(f"US MARKET: {len(us_strategies)} strategies @ ${us_capital:,.0f}") log("=" * 60) # US fee schedule (IBKR-style tiered): # fee = max(2.0, 1.88 + 0.009 * shares) us_raw = run_market( "us", us_strategies, years=args.years, capital=us_capital, tx_cost=0.001, fixed_fee=2.0, fee_base=1.88, fee_per_share=0.009, refresh=not args.offline, ) log("=" * 60) log(f"CN MARKET: {len(cn_strategies)} strategies @ ¥{cn_capital:,.0f}") log("=" * 60) cn_raw = run_market( "cn", cn_strategies, years=args.years, capital=cn_capital, tx_cost=0.001, fixed_fee=5.0, refresh=not args.offline, ) # Materialize every scenario × market by applying its injection schedule. scenarios_out: dict[str, dict] = {} for scenario in SCENARIOS: log(f"--- scenario: {scenario['id']} ---") scenarios_out[scenario["id"]] = { "id": scenario["id"], "label": scenario["label"], "description": scenario["description"], "capital": us_capital, "capitalByMarket": {"us": us_capital, "cn": cn_capital}, "injectionSchedule": _injection_for_market(scenario["injection"], us_capital), "markets": {}, } if us_raw: scenarios_out[scenario["id"]]["markets"]["us"] = build_scenario_for_market( us_raw, scenario, args.years, us_capital ) if cn_raw: scenarios_out[scenario["id"]]["markets"]["cn"] = build_scenario_for_market( cn_raw, scenario, args.years, cn_capital ) output = { "computedAt": datetime.now().isoformat(), "config": {"years": args.years, "initialCapital": MARKET_CAPITAL, "engineVersion": ENGINE_VERSION}, "scenarios": scenarios_out, } output_path = os.path.join(output_dir, "backtest_results.json") with open(output_path + ".tmp", "w") as f: json.dump(_sanitize_for_json(output), f, default=str) os.replace(output_path + ".tmp", output_path) file_size = os.path.getsize(output_path) / (1024 * 1024) log("=" * 60) log(f"DONE: Written to {output_path} ({file_size:.1f} MB)") for sid, sc in scenarios_out.items(): for mk, mv in sc["markets"].items(): log(f" [{sid}] {mk.upper()}: {len(mv['strategies'])} strategies") log("=" * 60) if __name__ == "__main__": main()