| |
| """Run strategy backtest on pre-mined factor signals.""" |
|
|
| import argparse |
| import json |
| import sys |
| from pathlib import Path |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| if str(ROOT) not in sys.path: |
| sys.path.insert(0, str(ROOT)) |
|
|
| import yaml |
|
|
| from config.settings import load_settings |
| from strategies.registry import load_strategy_catalog, list_strategies |
| from strategies.runner import run_strategy_backtest, run_strategy_suite, save_backtest_result |
|
|
|
|
| def _load_signal_source(name: str | None, path: str | None, config: dict | None) -> dict: |
| if config: |
| return config |
| catalog = load_strategy_catalog() |
| if name and name in catalog.get("signal_sources", {}): |
| return catalog["signal_sources"][name] |
| if path: |
| p = Path(path) |
| if p.suffix == ".json" and "factors" in p.read_text(encoding="utf-8")[:200]: |
| return {"type": "quantaalpha_library", "path": str(path)} |
| return {"type": "factor_panel", "path": str(path), "combine": "equal"} |
| raise ValueError("Provide --signal-config, --signal-source, or --signal-path") |
|
|
|
|
| def main(): |
| parser = argparse.ArgumentParser(description="Backtest pre-mined factors with pluggable strategies") |
| parser.add_argument("--strategy", type=str, default="topk_dropout", help="Strategy name") |
| parser.add_argument("--suite", type=str, default=None, help="Run preset strategy suite from config/strategies.yaml") |
| parser.add_argument("--signal-source", type=str, default=None, help="Named signal source from strategies.yaml") |
| parser.add_argument("--signal-path", type=str, default=None, help="Direct path to factor panel or QuantaAlpha JSON") |
| parser.add_argument("--signal-config", type=str, default=None, help="JSON signal source config") |
| parser.add_argument("--strategy-kwargs", type=str, default=None, help="JSON strategy kwargs override") |
| parser.add_argument("--start", type=str, default=None) |
| parser.add_argument("--end", type=str, default=None) |
| parser.add_argument("--output-dir", type=str, default=None) |
| parser.add_argument("--list-strategies", action="store_true") |
| args = parser.parse_args() |
|
|
| if args.list_strategies: |
| for s in list_strategies(): |
| print(f"- {s['name']}: {s.get('description', '')}") |
| return |
|
|
| signal_cfg = _load_signal_source( |
| args.signal_source, |
| args.signal_path, |
| json.loads(args.signal_config) if args.signal_config else None, |
| ) |
| strat_kwargs = json.loads(args.strategy_kwargs) if args.strategy_kwargs else None |
| settings = load_settings() |
| out_dir = Path(args.output_dir) if args.output_dir else settings.output_root / "backtest" |
|
|
| if args.suite: |
| catalog = load_strategy_catalog() |
| names = catalog.get("suites", {}).get(args.suite, []) |
| if not names: |
| raise ValueError(f"Unknown suite: {args.suite}") |
| results = run_strategy_suite(signal_cfg, names, start_time=args.start, end_time=args.end) |
| for name, res in results.items(): |
| save_backtest_result(res, out_dir / name) |
| print(f"[{name}] saved to {out_dir / name}") |
| return |
|
|
| result = run_strategy_backtest( |
| strategy_name=args.strategy, |
| signal_source=signal_cfg, |
| strategy_kwargs=strat_kwargs, |
| start_time=args.start, |
| end_time=args.end, |
| ) |
| save_backtest_result(result, out_dir / args.strategy) |
| print(f"Strategy={args.strategy}") |
| print(f"Signal stats: {result.signal_stats}") |
| if not result.risk.empty: |
| print(result.risk) |
| print(f"Saved to {out_dir / args.strategy}") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
