sfgrid/core/scoring/cli.py

"""
grid_seeker v6.4 CLI 入口

Usage:
    python -m core.scoring.cli sync all         # 按依赖顺序执行全部同步
    python -m core.scoring.cli sync kline       # 仅同步个股+指数K线
    python -m core.scoring.cli sync stocks      # 仅同步股票基础信息
    python -m core.scoring.cli sync industry    # 仅同步行业映射
    python -m core.scoring.cli sync market      # 仅计算市场状态
    python -m core.scoring.cli sync sector      # 仅计算行业指数
    python -m core.scoring.cli score            # 完整评分管道 (最新交易日)
    python -m core.scoring.cli score --date 20260615  # 指定日期
    python -m core.scoring.cli score --dry-run  # 试运行 (不写库)
    python -m core.scoring.cli check 000001     # 检查单只股票数据充分性
    python -m core.scoring.cli list-candidates  # 列出全部候选股
"""
import sys
from datetime import date, datetime


def main():
    if len(sys.argv) < 2:
        _usage()
        return

    cmd = sys.argv[1]

    if cmd == 'sync':
        _cmd_sync()
    elif cmd == 'score':
        _cmd_score()
    elif cmd == 'check':
        _cmd_check()
    elif cmd == 'list-candidates':
        _cmd_list_candidates()
    else:
        print(f'未知命令: {cmd}')
        _usage()


def _usage():
    print(__doc__)


# ============================================================
# sync 命令
# ============================================================
def _cmd_sync():
    target = sys.argv[2] if len(sys.argv) > 2 else 'all'
    from core.scoring.sync import (
        KlineStockSync, KlineIndexSync,
        StocksSync, IndustrySync,
        MarketRegimeSync, SectorFeaturesSync,
    )

    syncs = {
        'kline': [KlineStockSync, KlineIndexSync],
        'stocks': [StocksSync],
        'industry': [IndustrySync],
        'market': [MarketRegimeSync],
        'sector': [SectorFeaturesSync],
    }

    if target == 'all':
        # 按依赖顺序执行
        order = [
            ('K线(个股)', KlineStockSync(count=300)),
            ('K线(指数)', KlineIndexSync(count=300)),
            ('股票信息', StocksSync()),
            ('行业映射', IndustrySync()),
            ('市场状态', MarketRegimeSync()),
            ('行业指数', SectorFeaturesSync()),
        ]
        for label, sync in order:
            print(f'\n===== {label} =====')
            sync.run()
        print('\n===== 全部同步完成 =====')
    elif target in syncs:
        for cls in syncs[target]:
            cls().run()
    else:
        print(f'未知同步目标: {target}')
        print(f'可用: all, {", ".join(syncs.keys())}')


# ============================================================
# score 命令
# ============================================================
def _cmd_score():
    dry_run = '--dry-run' in sys.argv
    date_str = None
    for i, arg in enumerate(sys.argv):
        if arg == '--date' and i + 1 < len(sys.argv):
            date_str = sys.argv[i + 1]
            break

    if date_str:
        trade_date = datetime.strptime(date_str, '%Y%m%d').date()
    else:
        trade_date = date.today()

    print(f'评分日期: {trade_date}')

    from core.scoring.inference.scorer import GridSeekerPipeline

    engine = GridSeekerPipeline()
    rankings = engine.run(trade_date)

    if rankings.empty:
        print('无评分结果')
        return

    if not dry_run:
        engine.persist(rankings, trade_date)
        print(f'结果已写入 ScoringResult ({len(rankings)} 条)')

    # 打印 Top-20
    print('\n===== Top-20 =====')
    print(f'{"Rank":<6} {"Code":<10} {"Profit":>10} {"Rounds":>10} {"Prob":>10}')
    print('-' * 50)
    for code, row in rankings.head(20).iterrows():
        print(f'{int(row["score_rank"]):<6} {code:<10} '
              f'{row["stacking_probability"]:>10.4f} '
              f'{row.get("rank_predicted_rounds", 0):>10.2f} '
              f'{row.get("stacking_probability", 0):>10.4f}')


# ============================================================
# check 命令
# ============================================================
def _cmd_check():
    if len(sys.argv) < 3:
        print('Usage: python -m core.scoring.cli check <stock_code>')
        return

    stock_code = sys.argv[2]
    from core.scoring.features.validator import _check_kline_sufficiency

    ok, reason, close = _check_kline_sufficiency(stock_code, date.today())
    if ok:
        print(f'{stock_code}: ✅ 通过 (close={close:.2f})')
    else:
        print(f'{stock_code}: ❌ {reason}')


# ============================================================
# list-candidates 命令
# ============================================================
def _cmd_list_candidates():
    from core.scoring.features.validator import load_candidates

    ctx = load_candidates(date.today())
    print(f'候选股总数: {len(ctx.candidates)}')
    print(f'排除: {len(ctx.excluded)}')
    print(f'\n候选股 (前100):')
    for i, code in enumerate(ctx.candidates[:100]):
        print(f'  {i + 1}. {code}')
    if ctx.excluded:
        print(f'\n排除原因 (前20):')
        for i, (code, reason) in enumerate(list(ctx.excluded.items())[:20]):
            print(f'  {code}: {reason}')


if __name__ == '__main__':
    main()