#!/usr/bin/env python3 """ Factor formula registry CLI. Record formulas in config/factor_registry.yaml, then: - compute: auto-calc via qlib - backtest: single-factor IC + strategy backtest - add: append a new formula to the registry file """ 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)) from factor_engine.formula_registry import ( add_factor_to_registry, build_combined_panel, compute_all_factors, compute_factor, export_gp_seed_formulas, list_factor_specs, registry_output_dir, ) from factor_engine.operator_builder import ( build_expression, get_operator_catalog, list_built_factors, register_all_built_factors, register_built_factor, ) from factor_engine.single_factor_backtest import ( analyze_single_factor, backtest_single_factor, run_all_single_factor_backtests, ) def cmd_list(args): specs = list_factor_specs(enabled_only=args.enabled_only, tag=args.tag) if not specs: print("No factors in registry.") return for s in specs: flag = "✓" if s.enabled else "✗" tags = ",".join(s.tags or []) print(f"[{flag}] {s.name}") print(f" expr: {s.expression}") if s.description: print(f" desc: {s.description}") if tags: print(f" tags: {tags}") def cmd_compute(args): if args.all: compute_all_factors( enabled_only=not args.include_disabled, start_time=args.start, end_time=args.end, cache=not args.no_cache, ) print(f"All factors cached under: {registry_output_dir()}") return if not args.name: raise SystemExit("Provide --name or --all") s = compute_factor(args.name, start_time=args.start, end_time=args.end, cache=not args.no_cache) print(f"Computed {args.name}: {s.notna().sum()} values -> {registry_output_dir() / args.name}.parquet") def cmd_analyze(args): res = analyze_single_factor(args.name, segment=args.segment, start_time=args.start, end_time=args.end) print(json.dumps({**res, "ic_series": None, "quantile_spread": None}, ensure_ascii=False, indent=2, default=str)) print(f"\nIC metrics: {res['metrics']}") def cmd_backtest(args): if args.all: run_all_single_factor_backtests(strategy_name=args.strategy, segment=args.segment) return if not args.name: raise SystemExit("Provide --name or --all") res = backtest_single_factor( args.name, strategy_name=args.strategy, segment=args.segment, start_time=args.start, end_time=args.end, ) print(json.dumps(res, ensure_ascii=False, indent=2, default=str)) def cmd_add(args): spec = add_factor_to_registry( name=args.name, expression=args.expression, description=args.description or "", tags=[t.strip() for t in args.tags.split(",")] if args.tags else None, ) print(f"Added factor: {spec.name}") print(f" expression: {spec.expression}") def cmd_export_panel(args): panel = build_combined_panel( factor_names=args.names.split(",") if args.names else None, enabled_only=not args.include_disabled, ) out = registry_output_dir() / "combined_panel.parquet" panel.to_parquet(out, index=False) print(f"Combined panel: {panel.shape} -> {out}") def cmd_export_seeds(args): out = export_gp_seed_formulas() print(f"GP seed formulas exported: {out}") def cmd_build(args): if args.all: exprs = register_all_built_factors(enabled_only=not args.include_disabled) for name, expr in exprs.items(): print(f"[{name}] {expr}") print(f"Registered {len(exprs)} built factors -> config/factor_registry.yaml") return if not args.name: raise SystemExit("Provide --name or --all") if args.register: expr = register_built_factor(args.name) print(f"Registered {args.name} -> {expr}") else: expr = build_expression(args.name) print(f"{args.name}: {expr}") def cmd_operators(args): catalog = get_operator_catalog() print("Operator catalog:") for k, v in catalog.items(): print(f" {k}: {v}") print("\nBuilt factors:") for spec in list_built_factors(enabled_only=args.enabled_only): flag = "✓" if spec.enabled else "✗" print(f" [{flag}] {spec.name} — {spec.description}") def main(): parser = argparse.ArgumentParser(description="Factor formula registry: compute, analyze, backtest") sub = parser.add_subparsers(dest="command", required=True) p_list = sub.add_parser("list", help="List registered factors") p_list.add_argument("--enabled-only", action="store_true") p_list.add_argument("--tag", type=str, default=None) p_list.set_defaults(func=cmd_list) p_compute = sub.add_parser("compute", help="Compute factor(s) via qlib") p_compute.add_argument("--name", type=str, default=None) p_compute.add_argument("--all", action="store_true") p_compute.add_argument("--start", type=str, default=None) p_compute.add_argument("--end", type=str, default=None) p_compute.add_argument("--no-cache", action="store_true") p_compute.add_argument("--include-disabled", action="store_true") p_compute.set_defaults(func=cmd_compute) p_analyze = sub.add_parser("analyze", help="IC analysis for one factor") p_analyze.add_argument("--name", type=str, required=True) p_analyze.add_argument("--segment", type=str, default="test") p_analyze.add_argument("--start", type=str, default=None) p_analyze.add_argument("--end", type=str, default=None) p_analyze.set_defaults(func=cmd_analyze) p_bt = sub.add_parser("backtest", help="Single-factor IC + strategy backtest") p_bt.add_argument("--name", type=str, default=None) p_bt.add_argument("--all", action="store_true") p_bt.add_argument("--strategy", type=str, default="topk_dropout") p_bt.add_argument("--segment", type=str, default="test") p_bt.add_argument("--start", type=str, default=None) p_bt.add_argument("--end", type=str, default=None) p_bt.set_defaults(func=cmd_backtest) p_add = sub.add_parser("add", help="Add a new factor formula to registry YAML") p_add.add_argument("--name", type=str, required=True) p_add.add_argument("--expression", type=str, required=True) p_add.add_argument("--description", type=str, default="") p_add.add_argument("--tags", type=str, default="custom") p_add.set_defaults(func=cmd_add) p_panel = sub.add_parser("export-panel", help="Merge registry factors into one panel") p_panel.add_argument("--names", type=str, default=None, help="Comma-separated factor names") p_panel.add_argument("--include-disabled", action="store_true") p_panel.set_defaults(func=cmd_export_panel) p_seeds = sub.add_parser("export-seeds", help="Export formulas as GP mining seeds") p_seeds.set_defaults(func=cmd_export_seeds) p_build = sub.add_parser("build", help="Build qlib expression from operator tree") p_build.add_argument("--name", type=str, default=None) p_build.add_argument("--all", action="store_true", help="Register all built factors to factor_registry.yaml") p_build.add_argument("--register", action="store_true", help="Write compiled expression into factor_registry.yaml") p_build.add_argument("--include-disabled", action="store_true") p_build.set_defaults(func=cmd_build) p_ops = sub.add_parser("operators", help="Show operator catalog and built factor trees") p_ops.add_argument("--enabled-only", action="store_true") p_ops.set_defaults(func=cmd_operators) args = parser.parse_args() args.func(args) if __name__ == "__main__": main()