scripts/run_factor.py

#!/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()