repo_description: LLM驱动的 A/H/美股智能分析器：多数据源行情 + 实时新闻 + LLM决策仪表盘 + 多渠道推送，零成本定时运行，纯白嫖. LLM-powered stock analysis system for A/H/US markets.
repo_topics: agent ai aigc gemini llm quant quantitative-trading rag stock
code: # -*- coding: utf-8 -*-
  """
  ===================================
  交易日历模块 (Issue #373)
  ===================================
  
  职责：
  1. 按市场（A股/港股/美股）判断当日是否为交易日
  2. 按市场时区取“今日”日期，避免服务器 UTC 导致日期错误
  3. 支持 per-stock 过滤：只分析当日开市市场的股票
  
  依赖：exchange-calendars（可选，不可用时 fail-open）
  """
  
  import logging
  from datetime import date, datetime
  from typing import Optional, Set
  from zoneinfo import ZoneInfo
  
  logger = logging.getLogger(__name__)
  
  # Exchange-calendars availability
  _XCALS_AVAILABLE = False
  try:
      import exchange_calendars as xcals
      _XCALS_AVAILABLE = True
  except ImportError:
      logger.warning(
          "exchange-calendars not installed; trading day check disabled. "
          "Run: pip install exchange-calendars"
      )
  
  # Market -> exchange code (exchange-calendars)
  MARKET_EXCHANGE = {"cn": "XSHG", "hk": "XHKG", "us": "XNYS"}
  
  # Market -> IANA timezone for "today"
  MARKET_TIMEZONE = {
      "cn": "Asia/Shanghai",
      "hk": "Asia/Hong_Kong",
      "us": "America/New_York",
  }
  
  
  def get_market_for_stock(code: str) -> Optional[str]:
      """
      Infer market region for a stock code.
  
      Returns:
          'cn' | 'hk' | 'us' | None (None = unrecognized, fail-open: treat as open)
      """
      if not code or not isinstance(code, str):
          return None
      code = (code or "").strip().upper()
  
      from data_provider import is_us_stock_code, is_us_index_code, is_hk_stock_code
  
      if is_us_stock_code(code) or is_us_index_code(code):
          return "us"
      if is_hk_stock_code(code):
          return "hk"
      # A-share: 6-digit numeric
      if code.isdigit() and len(code) == 6:
          return "cn"
      return None
  
  
  def is_market_open(market: str, check_date: date) -> bool:
      """
      Check if the given market is open on the given date.
  
      Fail-open: returns True if exchange-calendars unavailable or date out of range.
  
      Args:
          market: 'cn' | 'hk' | 'us'
          check_date: Date to check
  
      Returns:
          True if trading day (or fail-open), False otherwise
      """
      if not _XCALS_AVAILABLE:
          return True
      ex = MARKET_EXCHANGE.get(market)
      if not ex:
          return True
      try:
          cal = xcals.get_calendar(ex)
          session = datetime(check_date.year, check_date.month, check_date.day)
          return cal.is_session(session)
      except Exception as e:
          logger.warning("trading_calendar.is_market_open fail-open: %s", e)
          return True
  
  
  def get_market_now(
      market: Optional[str], current_time: Optional[datetime] = None
  ) -> datetime:
      """
      Return current time in the market's local timezone.
  
      If current_time is naive, treat it as already expressed in the market timezone.
      Unknown markets fall back to the given datetime (or local system time).
      """
      tz_name = MARKET_TIMEZONE.get(market or "")
  
      if current_time is None:
          if tz_name:
              return datetime.now(ZoneInfo(tz_name))
          return datetime.now()
  
      if not tz_name:
          return current_time
  
      tz = ZoneInfo(tz_name)
      if current_time.tzinfo is None:
          return current_time.replace(tzinfo=tz)
      return current_time.astimezone(tz)
  
  
  def get_effective_trading_date(
      market: Optional[str], current_time: Optional[datetime] = None
  ) -> date:
      """
      Resolve the latest reusable daily-bar date for checkpoint/resume logic.
  
      Rules:
      - Non-trading day / holiday: previous trading session
      - Trading day before market close: previous completed trading session
      - Trading day after market close: current trading session
      - Calendar lookup failure: fail-open to market-local natural date
      """
      market_now = get_market_now(market, current_time=current_time)
      fallback_date = market_now.date()
  
      if not _XCALS_AVAILABLE:
          return fallback_date
  
      ex = MARKET_EXCHANGE.get(market or "")
      tz_name = MARKET_TIMEZONE.get(market or "")
      if not ex or not tz_name:
          return fallback_date
  
      try:
          cal = xcals.get_calendar(ex)
          local_date = market_now.date()
  
          if not cal.is_session(local_date):
              return cal.date_to_session(local_date, direction="previous").date()
  
          session = cal.date_to_session(local_date, direction="previous")
          session_close = cal.session_close(session)
          if hasattr(session_close, "tz_convert"):
              close_local = session_close.tz_convert(tz_name).to_pydatetime()
          elif session_close.tzinfo is not None:
              close_local = session_close.astimezone(ZoneInfo(tz_name))
          else:
              close_local = session_close.replace(tzinfo=ZoneInfo(tz_name))
  
          if market_now >= close_local:
              return session.date()
  
          return cal.previous_session(session).date()
      except Exception as e:
          logger.warning("trading_calendar.get_effective_trading_date fail-open: %s", e)
          return fallback_date
  
  
  def get_open_markets_today() -> Set[str]:
      """
      Get markets that are open today (by each market's local timezone).
  
      Returns:
          Set of market keys ('cn', 'hk', 'us') that are trading today
      """
      if not _XCALS_AVAILABLE:
          return {"cn", "hk", "us"}
      result: Set[str] = set()
      for mkt, tz_name in MARKET_TIMEZONE.items():
          try:
              tz = ZoneInfo(tz_name)
              today = datetime.now(tz).date()
              if is_market_open(mkt, today):
                  result.add(mkt)
          except Exception as e:
              logger.warning("get_open_markets_today fail-open for %s: %s", mkt, e)
              result.add(mkt)
      return result
  
  
  def compute_effective_region(
      config_region: str, open_markets: Set[str]
  ) -> Optional[str]:
      """
      Compute effective market review region given config and open markets.
  
      Args:
          config_region: From MARKET_REVIEW_REGION ('cn' | 'us' | 'both')
          open_markets: Markets open today
  
      Returns:
          None: caller uses config default (check disabled)
          '': all relevant markets closed, skip market review
          'cn' | 'us' | 'both': effective subset for today
      """
      if config_region not in ("cn", "us", "both"):
          config_region = "cn"
      if config_region == "cn":
          return "cn" if "cn" in open_markets else ""
      if config_region == "us":
          return "us" if "us" in open_markets else ""
      # both
      parts = []
      if "cn" in open_markets:
          parts.append("cn")
      if "us" in open_markets:
          parts.append("us")
      if not parts:
          return ""
      return "both" if len(parts) == 2 else parts[0]
  