qlib/scripts/run_dividend_case.py

"""高股息选股回测案例执行器（走真实 Job 链路：submit → 状态机 → Experiment 归档）。

案例口径（全部可改，见 --help）：
- 全市场股息率最高的 n 只（n=20），自 2020-01-01 起每 m 个月（m=6）择股一次
- 附加条件：dv_ratio ≤ 30%（剔除特殊分红导致的股息率尖峰）
- 实际持仓前 x 只（x=20，等权）；每 y 个月（y=6）调仓，买卖点为收盘价
- 起始资金 100 万；佣金 0.03% / 印花税 0.05% / 滑点 0.1% / 最低佣金 5 元
- 行情口径 hfq（后复权）：现金分红不再被计为除权亏损
- 买不进（涨停/停牌）→ 顺延到之后首个可成交交易日按收盘价买入

用法（在 backend/ 下执行，与 Job 子进程同一环境）：
    PYTHONPATH=. .venv/bin/python ../scripts/run_dividend_case.py
    PYTHONPATH=. .venv/bin/python ../scripts/run_dividend_case.py --n 30 --x 10 --end 2025-12-31

执行成功后打印关键指标，并把完整结果（含净值曲线 / 个股曲线 / 买卖点）落盘为 JSON：
    data/backtest_dividend_case.json
"""

from __future__ import annotations

import argparse
import json
from datetime import date
from pathlib import Path

from app.application.services.job_executor import submit_and_run
from app.domain.entities.research import (
    ConditionSpec,
    CostSpec,
    FactorSpec,
    ResearchSpec,
    SelectionSpec,
    UniverseSpec,
)


def build_spec(args: argparse.Namespace) -> ResearchSpec:
    return ResearchSpec(
        type="backtest",
        universe=UniverseSpec(exclude_st=not args.include_st, min_listing_days=args.min_listing_days),
        price_adjustment=args.adjust,
        factors=[FactorSpec(name=args.factor, weight=1.0)],
        conditions=[ConditionSpec(field="dv_ratio", op="lte", value=args.dv_cap)],
        selection=SelectionSpec(
            top_n=args.n,
            hold_top_x=args.x,
            allow_substitute=False,
            defer_buy=not args.no_defer,
        ),
        rebalance="monthly",
        selection_interval_months=args.m,
        rebalance_interval_months=args.y if args.y is not None else args.m,
        period=(args.start, args.end),
        costs=CostSpec(
            commission_rate=args.commission,
            stamp_tax_rate=args.stamp,
            slippage_rate=args.slippage,
            min_commission=args.min_commission,
        ),
        initial_capital=args.capital,
    )


def main() -> int:
    p = argparse.ArgumentParser(description="高股息选股回测案例（真实 Job 链路）")
    p.add_argument("--n", type=int, default=20, help="候选池大小 n（股息率最高的 n 只）")
    p.add_argument("--x", type=int, default=20, help="实际持仓数 x（必须 ≤ n）")
    p.add_argument("--m", type=int, default=6, help="择股间隔（月）")
    p.add_argument("--y", type=int, default=None, help="调仓间隔（月）；缺省 = m")
    p.add_argument("--start", type=date.fromisoformat, default=date(2020, 1, 1))
    p.add_argument("--end", type=date.fromisoformat, default=date.today())
    p.add_argument("--factor", default="dividend_yield", help="因子名（默认股息率 dv_ratio）")
    p.add_argument("--dv-cap", type=float, default=30.0, help="股息率上限（%%），剔除尖峰")
    p.add_argument(
        "--adjust", choices=["none", "qfq", "hfq"], default="hfq", help="复权口径"
    )
    p.add_argument("--capital", type=float, default=1_000_000.0)
    p.add_argument("--commission", type=float, default=0.0003)
    p.add_argument("--stamp", type=float, default=0.0005)
    p.add_argument("--slippage", type=float, default=0.001)
    p.add_argument("--min-commission", type=float, default=5.0, help="单笔最低佣金（元）")
    p.add_argument("--min-listing-days", type=int, default=250)
    p.add_argument("--include-st", action="store_true", help="不剔除 ST（默认剔除）")
    p.add_argument("--no-defer", action="store_true", help="买不进时不顺延（直接放弃）")
    p.add_argument(
        "--out",
        type=Path,
        default=Path("../data/backtest_dividend_case.json"),
        help="结果 JSON 落盘路径",
    )
    args = p.parse_args()
    if args.x > args.n:
        p.error(f"--x({args.x}) 不能大于 --n({args.n})")

    spec = build_spec(args)
    print("[case] 已提交 Job（真实状态机 + Experiment 归档）…", flush=True)
    job = submit_and_run(spec)
    print(f"[case] job={job.id} status={job.status} mode/kind={job.kind}")
    if job.status != "success":
        print(f"[case] 失败：{job.error}")
        return 1

    result = json.loads(job.result_json or "{}")
    s = result.get("summary", {})
    print("\n=== 回测结果（整体收益趋势）===")
    for key in (
        "start", "end", "initial_capital", "final_equity", "total_return_pct",
        "annual_return_pct", "sharpe", "max_drawdown_pct", "volatility_pct",
        "win_rate_pct", "total_trades", "avg_turnover_pct",
    ):
        print(f"  {key:22s} {s.get(key)}")
    curves = result.get("symbol_curves", [])
    print(f"\n=== 个股收益率趋势（共 {len(curves)} 只，按期末收益绝对值降序，前 10）===")
    for c in curves[:10]:
        buys = sum(1 for m in c.get("marks", []) if m.get("signal") == "BUY")
        sells = sum(1 for m in c.get("marks", []) if m.get("signal") == "SELL")
        print(f"  {c['symbol']}  期末 {c['final_return_pct']:8.2f}%  买点 {buys}  卖点 {sells}")
    print("\n=== 未建模 / 口径说明（如实标注）===")
    for note in result.get("unimplemented", []):
        print(f"  - {note}")
    print("\n=== 买卖点（signal_history 中的成交，前 10 条）===")
    for f in [a for a in result.get("fills", [])][:10]:
        print(f"  {f['date']} {f['symbol']} {f['signal']} @ {f.get('price')}")

    out = args.out
    out.parent.mkdir(parents=True, exist_ok=True)
    out.write_text(json.dumps(result, ensure_ascii=False), encoding="utf-8")
    print(f"\n[case] 完整结果已写入 {out.resolve()}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())