myquant/finance/cli/agent_cli.py

#!/usr/bin/env python
"""
Agent 命令行入口。

用法:
    python cli/agent_cli.py daily          # 执行每日流程
    python cli/agent_cli.py picks [N]      # 今日选股 Top N
    python cli/agent_cli.py risk           # 风险评估
    python cli/agent_cli.py research       # 因子研究
    python cli/agent_cli.py report [DATE]  # 生成日报
"""

import sys
import os
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from datetime import datetime


def init_engines():
    """初始化所有引擎。"""
    from data.data_manager import DataManager
    from factors.engine import FactorEngine
    from backtest.vectorbt.engine import VectorBTEngine
    from optimizer.engine import OptunaEngine
    from factors.sentiment.sentiment_engine import SentimentEngine
    from factors.sentiment.news_source import NewsSource
    from factors.sentiment.qwen_client import QwenClient
    from factors.registry import get_factor

    dm = DataManager()
    dm.init_db()
    fe = FactorEngine(dm)
    bt = VectorBTEngine()
    opt = OptunaEngine(bt)
    sent = SentimentEngine(dm, qwen_client=QwenClient(), news_source=NewsSource())
    fe._sentiment_engine = sent

    return {
        "dm": dm,
        "fe": fe,
        "bt": bt,
        "opt": opt,
        "sent": sent,
    }


def main():
    if len(sys.argv) < 2:
        print("用法: agent_cli.py <daily|picks|risk|research|report|warmup>")
        print()
        print("  daily [DATE]    — 执行每日完整流程")
        print("  picks [N] [DATE] — 今日选股 Top N")
        print("  risk            — 风险评估")
        print("  research        — 因子发现与评估")
        print("  report [DATE]   — 生成日报")
        print("  warmup [N]      — 首次批量预热范围股票到 DB 缓存")
        return

    cmd = sys.argv[1]
    engines = init_engines()

    from agents.orchestrator import AgentOrchestrator
    orch = AgentOrchestrator(**engines)
    orch.setup()

    if cmd == "daily":
        date = sys.argv[2] if len(sys.argv) > 2 else None
        results = orch.run_daily(date=date)
        # 打印日报内容
        report_md = results.get("report", {}).get("report_markdown", "")
        if report_md:
            print(report_md)

    elif cmd == "picks":
        n = int(sys.argv[2]) if len(sys.argv) > 2 else 15
        date = sys.argv[3] if len(sys.argv) > 3 else None
        result = orch.picks(date=date, top_n=n)
        print(f"\n选股结果 ({result.get('date', '?')}):")
        for p in result.get("top_picks", []):
            print(f"  {p['ts_code']:12s} {p.get('name', ''):10s} {p['score']:.4f}")

    elif cmd == "risk":
        result = orch.risk_check()
        print(f"\n风险评估:")
        print(f"  等级: {result['risk_level']}")
        print(f"  建议仓位: {result['target_exposure']:.0%}")
        print(f"  止损线: {result['stop_loss']:.0%}")
        print(f"  单票上限: {result['max_single_position']:.0%}")
        indicators = result.get("indicators", {})
        if indicators:
            print(f"  波动率: {indicators.get('market_volatility', 0):.1f}%")
            print(f"  回撤: {indicators.get('current_drawdown', 0):.1f}%")
        for a in result.get("alerts", []):
            print(f"  ⚠️ {a}")

    elif cmd == "research":
        result = orch.run_research_cycle()
        top = result.get("research", {}).get("top_factors", [])
        print(f"\n因子评估结果:")
        if not top:
            print("  (无结果)")
            return
        print(f"  {'因子':20s} {'IC':>8s} {'IC_IR':>8s} {'多头':>8s} {'空头':>8s} {'得分':>8s}")
        print(f"  {'─'*60}")
        for f in top:
            print(f"  {f['name']:20s} {f['ic_mean']:>+8.4f} {f['icir']:>8.3f} "
                  f"{f['long_ret']:>+7.1f}% {f['short_ret']:>+7.1f}% {f['score']:>8.4f}")

    elif cmd == "warmup":
        batch_n = int(sys.argv[2]) if len(sys.argv) > 2 else 50
        print("首次批量预热: 每次 {} 只股票，分批执行...".format(batch_n))
        sent = engines.get("sent")
        dm = engines.get("dm")
        scope = sent.get_scope_stocks() if sent else list(dm.get_stock_list().index[:100])
        from database.dao import get_latest_trade_date
        uncached = [c for c in scope if not get_latest_trade_date(c)]
        print("范围: {} 只, 未缓存: {} 只".format(len(scope), len(uncached)))

        total_synced = 0
        for i in range(0, len(uncached), batch_n):
            batch = uncached[i:i + batch_n]
            print("[warmup] 批次 {}/{} ({}~{})".format(i // batch_n + 1, (len(uncached) - 1) // batch_n + 1, i, i + len(batch)))
            for ts_code in batch:
                try:
                    n = dm.sync_daily(ts_code)
                    total_synced += n
                except Exception as e:
                    print("  {} 失败: {}".format(ts_code, e))
            print("  累计同步: {} 条".format(total_synced))
        print("预热完成: {} 条数据, {} 只新股票已缓存".format(total_synced, len(uncached)))

    elif cmd == "report":
        date = sys.argv[2] if len(sys.argv) > 2 else None
        result = orch.generate_report(date=date)
        print("\n日报已生成: {}".format(result.get("report_path", "?")))
        # 存入 DB
        md = result.get("report_markdown", "")
        if md:
            from reports.storage import save_report
            save_report(md, "量化日报", report_date=date or datetime.now().strftime("%Y%m%d"),
                        subject_type="daily", subject_code="")
            print("  已存入 DB")
        if md:
            print(md)

    else:
        print(f"未知命令: {cmd}")


if __name__ == "__main__":
    main()