File size: 3,659 Bytes
590a501 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 | #!/usr/bin/env python3
"""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()
|