Qionk/a-share-quant

1#!/usr/bin/env python32"""3A股量化辅助决策模型 - CLI 入口4 5用法:6    python run.py update        更新行情数据（首次较慢）7    python run.py signal        生成今日关注池8    python run.py backtest      运行策略回测9    python run.py evaluate      因子 IC 评估10    python run.py all           依次执行全部11"""12 13import os14import sys15 16# 确保项目根目录在 sys.path 中17ROOT = os.path.dirname(os.path.abspath(__file__))18sys.path.insert(0, ROOT)19os.chdir(ROOT)20 21from src.data import load_config, update_all_data, get_stock_pool, load_panel_data22from src.factors import compute_all_factors, evaluate_factors23from src.signal import generate_signals24from src.backtest import backtest, calc_metrics25 26 27def _load_pipeline(config):28    """公共流程：加载数据 → 因子 → 信号"""29    pool = get_stock_pool(config)30    print(f"股票池: {len(pool)} 只")31    panel = load_panel_data(config, codes=pool["code"].tolist())32    if not panel:33        print("无数据，请先运行: python run.py update")34        sys.exit(1)35    factors, breadth = compute_all_factors(panel, config)36    signals, scores, regime = generate_signals(factors, breadth, config)37    return pool, panel, factors, breadth, signals, scores, regime38 39 40# ── 命令 ─────────────────────────────────────────────────────41 42 43def cmd_update(config):44    print("=" * 50)45    print("  数据更新")46    print("=" * 50)47    update_all_data(config)48 49 50def cmd_signal(config):51    print("=" * 50)52    print("  生成今日信号")53    print("=" * 50)54    _, panel, factors, breadth, signals, scores, regime = _load_pipeline(config)55 56    latest = scores.index[-1]57    print(f"\n日期: {latest.strftime('%Y-%m-%d')}")58    print(f"市场宽度: {breadth.iloc[-1]:.1%}")59    print(f"市场状态: {regime.iloc[-1]}")60 61    today_scores = scores.loc[latest].dropna().sort_values(ascending=False)62    top_n = config["signal"]["top_n"]63    top = today_scores.head(top_n)64 65    print(f"\nTop {top_n} 关注股票:")66    print("-" * 60)67    print(f"  {'#':>3}  {'代码':<8} {'评分':>8} {'相对强度':>8} {'趋势':>6} {'确认':>4}")68    print("-" * 60)69    for i, (code, score) in enumerate(top.items(), 1):70        rs = factors["relative_strength"].loc[latest].get(code, 0)71        trend = factors["trend"].loc[latest].get(code, 0)72        confirmed = "是" if signals.loc[latest].get(code, False) else "否"73        print(f"  {i:3d}  {code:<8} {score:8.3f} {rs:8.2f} {trend:6.2f} {confirmed:>4}")74 75    os.makedirs("output", exist_ok=True)76    import pandas as pd77    out = pd.DataFrame({78        "code": top.index,79        "score": top.values,80        "relative_strength": [factors["relative_strength"].loc[latest].get(c, 0) for c in top.index],81        "trend": [factors["trend"].loc[latest].get(c, 0) for c in top.index],82        "confirmed": [signals.loc[latest].get(c, False) for c in top.index],83    })84    out.to_csv("output/signals.csv", index=False)85    print(f"\n信号已保存到 output/signals.csv")86 87 88def cmd_backtest(config):89    print("=" * 50)90    print("  策略回测")91    print("=" * 50)92    _, panel, factors, breadth, signals, scores, regime = _load_pipeline(config)93 94    results = backtest(signals, scores, panel["close"], config)95 96    if not results["daily_returns"].empty:97        metrics = calc_metrics(results["daily_returns"])98        print("\n回测结果:")99        print("-" * 40)100        for k, v in metrics.items():101            print(f"  {k}: {v}")102 103        os.makedirs("output", exist_ok=True)104        results["daily_returns"].to_csv("output/backtest_result.csv")105        if not results["trade_log"].empty:106            results["trade_log"].to_csv("output/trade_log.csv", index=False)107        print("\n结果已保存到 output/")108    else:109        print("回测数据不足")110 111 112def cmd_evaluate(config):113    print("=" * 50)114    print("  因子 IC 评估")115    print("=" * 50)116    _, panel, factors, breadth, _, _, _ = _load_pipeline(config)117 118    ic_results = evaluate_factors(factors, panel["close"])119    print(f"\n{'因子':<20} {'IC均值':>10} {'ICIR':>10} {'IC>0':>10}")120    print("-" * 55)121    for name, r in ic_results.items():122        print(f"  {name:<18} {r['ic_mean']:+10.4f} {r['icir']:+10.4f} {r['ic_positive_pct']:10.1%}")123 124 125# ── main ─────────────────────────────────────────────────────126 127 128def main():129    config = load_config()130 131    if len(sys.argv) < 2:132        print(__doc__)133        return134 135    cmd = sys.argv[1]136    dispatch = {137        "update": cmd_update,138        "signal": cmd_signal,139        "backtest": cmd_backtest,140        "evaluate": cmd_evaluate,141    }142 143    if cmd == "all":144        for fn in [cmd_update, cmd_signal, cmd_evaluate, cmd_backtest]:145            fn(config)146            print()147    elif cmd in dispatch:148        dispatch[cmd](config)149    else:150        print(f"未知命令: {cmd}")151        print(__doc__)152 153 154if __name__ == "__main__":155    main()156