us_main.py

"""Matrix 美股推荐系统入口程序。

与 ETF 的 main.py、A 股的 stock_main.py **完全独立**：使用独立的美股数据库
（``data/matrix_us.db``）、独立标的池（``US_Equity``）和独立的美股策略集，互不影响。

运行模式：
  python us_main.py                    # 日常模式：增量同步 + 跑美股策略 + 飞书推送
  python us_main.py --backfill         # 回填模式：拉取 US_Equity 全量历史日 K（首次使用）
  python us_main.py --sync-universe    # 仅同步美股标的池与基础信息（stock_basic）
  python us_main.py --symbols AAPL.US,MSFT.US   # 仅处理指定美股
  python us_main.py --force            # 非交易日也强制运行
"""

import argparse
import os
import socket
import sys
from datetime import date
from pathlib import Path

from dotenv import load_dotenv

PROJECT_ROOT = Path(__file__).resolve().parent
os.chdir(PROJECT_ROOT)
load_dotenv(PROJECT_ROOT / ".env")

socket.setdefaulttimeout(30.0)

from matrix_etf.analytics.integration import AnalyticsHook  # noqa: E402
from matrix_etf.core.config import get_settings  # noqa: E402
from matrix_etf.core.logger import get_logger  # noqa: E402
from matrix_etf.core.trading_calendar import get_us_non_trading_day_reason  # noqa: E402
from matrix_etf.data.sync_runner import sync_until_stable  # noqa: E402
from matrix_etf.data.us_stock_engine import UsStockDataEngine  # noqa: E402
from matrix_etf.notify.feishu import FeishuNotifier  # noqa: E402
from matrix_etf.strategy.base import BaseStrategy  # noqa: E402
from matrix_etf.strategy.ranking import select_recommendations  # noqa: E402
from matrix_etf.strategy.us.breakout_volume import UsBreakoutVolumeStrategy  # noqa: E402
from matrix_etf.strategy.us.ma_volume import UsMaVolumeStrategy  # noqa: E402
from matrix_etf.strategy.us.rps_momentum import UsRpsMomentumStrategy  # noqa: E402
from matrix_etf.strategy.us.trend_ma import UsTrendMaStrategy  # noqa: E402


def _parse_symbols(value: str | None) -> list[str] | None:
    if not value:
        return None
    symbols = [item.strip() for item in value.split(",") if item.strip()]
    return symbols or None


def _build_strategies(engine: UsStockDataEngine, settings) -> list[BaseStrategy]:
    return [
        UsRpsMomentumStrategy(engine=engine, settings=settings),
        UsTrendMaStrategy(engine=engine, settings=settings),
        UsMaVolumeStrategy(engine=engine, settings=settings),
        UsBreakoutVolumeStrategy(engine=engine, settings=settings),
    ]


def _run_backfill(engine: UsStockDataEngine, requested: list[str] | None, logger) -> None:
    logger.info("进入美股回填模式...")
    if requested:
        symbols = requested
        engine.sync_basic_info(symbols)
    else:
        symbols = engine.sync_universe_and_get_symbols()
    engine.backfill(symbols)
    logger.info("Matrix 美股回填模式运行完成（美股名称已随日 K 自动入库）")


def _run_explicit_command(
    args, engine: UsStockDataEngine, requested: list[str] | None, logger
) -> bool:
    commands = [
        (args.backfill, lambda: _run_backfill(engine, requested, logger)),
        (args.sync_universe, lambda: engine.sync_universe()),
    ]
    for enabled, runner in commands:
        if enabled:
            runner()
            return True
    return False


def main() -> None:
    parser = argparse.ArgumentParser(description="Matrix 美股推荐系统")
    parser.add_argument(
        "--backfill",
        action="store_true",
        help="回填模式：拉取 US_Equity 全量历史日 K（首次使用）",
    )
    parser.add_argument(
        "--sync-universe",
        action="store_true",
        help="仅同步美股标的池与基础信息（stock_basic）",
    )
    parser.add_argument(
        "--symbols",
        help="仅处理指定美股，多个代码用逗号分隔，例如 AAPL.US,MSFT.US",
    )
    parser.add_argument(
        "--force",
        action="store_true",
        help="日常模式下即使今天是周末或配置的美股休市日也继续运行",
    )
    args = parser.parse_args()

    try:
        settings = get_settings()

        logger = get_logger(__name__)
        logger.info("Matrix 美股启动")

        engine = UsStockDataEngine(settings)
        requested_symbols = _parse_symbols(args.symbols)

        if _run_explicit_command(args, engine, requested_symbols, logger):
            return

        # ── 日常模式：增量同步 + 跑美股策略 + 推送 ──
        if settings.skip_non_trading_day and not args.force:
            reason = get_us_non_trading_day_reason(date.today(), settings)
            if reason:
                logger.info(f"今日为非交易日（{reason}），跳过美股同步；如需运行请加 --force")
                return

        if requested_symbols:
            symbols = requested_symbols
            try:
                engine.sync_basic_info(requested_symbols)
            except Exception as exc:  # noqa: BLE001
                logger.warning(f"基础信息同步失败，将使用本地已有数据继续：{exc}")
        else:
            try:
                symbols = engine.sync_universe_and_get_symbols()
            except Exception as exc:  # noqa: BLE001
                symbols = engine.get_local_symbols()
                logger.warning(
                    f"标的池/基础信息同步失败，改用本地已有 {len(symbols)} 只美股继续：{exc}"
                )

        notifier = FeishuNotifier(settings, engine=engine)

        local_symbols = set(engine.get_local_symbols())
        has_local_data = any(symbol in local_symbols for symbol in symbols)
        if not has_local_data:
            logger.info("本地暂无美股数据，自动执行首次回填...")
            engine.backfill(symbols)
        else:
            logger.info("开始持续同步最新日 K，直到拉全或达标...")
            # 美股免费档为历史数据，且预期最新交易日在中国时区不易精确推断，
            # 故不传 expected_latest_date，仅按覆盖率收敛判定拉取完成。
            outcome = sync_until_stable(
                engine,
                symbols,
                expected_latest_date=None,
                max_seconds=settings.sync_persist_max_seconds,
                round_interval=settings.sync_persist_round_interval,
                target_coverage=settings.sync_persist_target_coverage,
                min_coverage=settings.sync_persist_min_coverage,
                log=logger,
            )
            if not outcome.success:
                notifier.send_alert(
                    message=(
                        f"⚠️ {date.today():%Y-%m-%d} 美股数据经持续重试后仍未拉全："
                        f"{outcome.describe()}。本次跳过策略推送，可能为数据源限流/网络"
                        "异常，请排查后手动重跑 `python us_main.py --force`。"
                    ),
                    category="US",
                )
                logger.warning("数据持续拉取失败，已发送告警卡片并跳过美股策略推送")
                return

        strategies = _build_strategies(engine, settings)
        analytics = AnalyticsHook(settings, market="US")

        candidates = []
        for strategy in strategies:
            strategy_name = type(strategy).__name__
            logger.info(f"执行策略：{strategy_name}")
            candidates.append(strategy.run())
        recommendations = select_recommendations(
            engine, candidates, settings.recommendation_limit
        )
        for strategy, selected in zip(strategies, recommendations):
            strategy_name = type(strategy).__name__
            logger.info(f"{strategy_name} 择优保留 {len(selected)} 只美股")
            if selected:
                perf_line = analytics.record_and_perf_line(strategy, selected)
                notifier.send(
                    symbols=selected,
                    strategy_name=strategy_name,
                    webhook_key=strategy.webhook_key,
                    category="US",
                    perf_line=perf_line,
                )
            else:
                logger.info(f"{strategy_name} 无选股结果，跳过推送")

    except Exception:
        try:
            _logger = get_logger(__name__)
            _logger.exception("美股主流程发生未捕获异常，程序终止")
        except Exception:
            import traceback
            traceback.print_exc()
        sys.exit(1)

    logger.info("Matrix 美股运行完成")


if __name__ == "__main__":
    main()