MoFin/deploy/profile-scripts/market_config.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""market_config.py — 市场抽象层（阶段1：港股接入核心结构件）

市场判断 / 行情符号 / 市场配置的**唯一事实源**，收敛散落在
market_data / price_monitor / technical_analysis /
strategy_lab 中的 6 处重复实现。

铁律：
  - 市场判断一律走 market_for_code（委托 mo_models.is_hk_stock，5位0/1开头=港股）
  - 行情符号一律走 kline_symbol（统一 sh/sz/hk 前缀规则）
  - 市场时段/币种/涨跌停配置一律从 MARKETS 读取
  - mo_models 不依赖本模块（无循环导入）；本模块只依赖 mo_models
"""

from datetime import datetime
from pathlib import Path

from mo_models import is_hk_stock


# ── 市场判断（唯一事实源：mo_models.is_hk_stock）────────────────────────

def market_for_code(code):
    """统一市场判断：返回 'a'（A股）| 'hk'（港股）。

    规则与 mo_models.is_hk_stock 完全一致：5位数字且0/1开头 → hk，其余 → a。
    6位A股、带前缀代码（sh/sz/hk/...）均正确归一。
    """
    return 'hk' if is_hk_stock(code) else 'a'


# ── 行情符号映射（收敛 market_data/price_monitor/technical_analysis 三处）──

def _normalize_code(code):
    """归一化代码：去市场前缀（sh/sz/hk/SH...）与 '_' 后缀，返回大写纯代码"""
    s = str(code or '').strip().upper()
    for p in ('HK', 'SH', 'SZ', 'BJ'):
        if s.startswith(p):
            s = s[len(p):]
    if '_' in s:
        s = s.split('_')[0]
    return s


def kline_symbol(code):
    """腾讯行情符号映射：'600519'→'sh600519', '000001'→'sz000001', '00700'→'hk00700'。

    统一规则（收敛两个来源的规则差异）：
      - 5位数字 → hk
      - 6位数字：5/6/9开头 → sh；0/1/2/3开头 → sz
        （4/7/8开头按 sz 兜底——现实中无此类A股代码，price_monitor/technical_analysis
         原实现即按 sz 处理）
    非5/6位数字（空、字母、超长）→ None。
    """
    raw = _normalize_code(code)
    if not raw or not raw.isdigit():
        return None
    if len(raw) == 5:
        return f"hk{raw}"
    if len(raw) == 6:
        if raw.startswith(('5', '6', '9')):
            return f"sh{raw}"
        return f"sz{raw}"
    return None


# ── 市场配置 ────────────────────────────────────────────────────────────

MARKETS = {
    'a': {
        'index_code': 'sh000001',       # A股大盘：上证指数
        'currency': 'CNY',
        'has_price_limit': True,
        'trading_hours': [('09:30', '11:30'), ('13:00', '15:00')],
    },
    'hk': {
        'index_code': 'hkHSI',          # 港股大盘：恒生指数
        'currency': 'HKD',
        'has_price_limit': False,
        'trading_hours': [('09:30', '12:00'), ('13:00', '16:00')],  # 港股午休12:00-13:00
    },
}


def market_config(market):
    """市场配置 dict：MARKETS[market]（market 为 'a' 或 'hk'）"""
    return MARKETS[market]


def is_trading_now(market, dt=None):
    """是否在交易时段内：工作日（weekday<5）且当前时间落在任一交易时段内。

    dt: datetime 对象，缺省取当前时间。
    """
    if market not in MARKETS:
        return False
    dt = dt or datetime.now()
    if dt.weekday() >= 5:
        return False
    t = dt.strftime('%H:%M')
    for start, end in MARKETS[market]['trading_hours']:
        if start <= t <= end:
            return True
    return False


def get_regime_temp(code):
    """按标的市场读取温区+温度（信号溯源用，2026-08-14 阶段4）。

    A股：smoothed 顶层（get_current_regime，K=5 平滑温区）+ get_market_temp('a')
        —— 与既有单市场行为完全一致。
    港股：smoothed markets.hk（K=5 平滑温区）+ get_market_temp('hk')。

    返回 (regime, temp_band)，异常时 ("", "")。延迟 import 避免循环依赖。
    """
    mkt = market_for_code(code)
    regime, temp = "", ""
    try:
        if mkt == 'hk':
            import json as _json
            sp = Path(__file__).resolve().parent.parent.parent / "data" / "market_regime_smoothed.json"
            if sp.exists():
                d = _json.loads(sp.read_text(encoding="utf-8"))
                mk = (d.get("markets") or {}).get("hk") or {}
                regime = mk.get("current_regime", "")
                temp = (mk.get("temp") or {}).get("band", "")
            if not regime:  # 回退：market_regime 表港股最新温区
                import sqlite3 as _sq
                from market_regime import DB_PATH as _DB
                conn = _sq.connect(str(_DB), timeout=5)
                row = conn.execute(
                    "SELECT regime FROM market_regime WHERE market='hk' ORDER BY date DESC LIMIT 1"
                ).fetchone()
                conn.close()
                regime = row[0] if row else ""
        else:
            from regime_gate import get_current_regime
            regime = get_current_regime().get("regime", "")
            from temp_band import get_market_temp
            temp = get_market_temp(market='a').get("band", "")
    except Exception:
        pass
    return regime, temp


# ── 模块自检 ────────────────────────────────────────────────────────────

if __name__ == '__main__':
    cases = [
        # (code, expected_market, expected_symbol)
        ('600519', 'a', 'sh600519'),
        ('000001', 'a', 'sz000001'),
        ('300750', 'a', 'sz300750'),
        ('688111', 'a', 'sh688111'),
        ('510050', 'a', 'sh510050'),
        ('002594', 'a', 'sz002594'),
        ('00700',  'hk', 'hk00700'),
        ('09988',  'hk', 'hk09988'),
        ('01810',  'hk', 'hk01810'),
        ('sh600519', 'a', 'sh600519'),
        ('hk00700',  'hk', 'hk00700'),
        ('AAPL',   'a', None),
        ('',       'a', None),
    ]
    print("=== market_for_code / kline_symbol 测试 ===")
    ok = True
    for code, exp_m, exp_s in cases:
        m = market_for_code(code)
        s = kline_symbol(code)
        m_ok = m == exp_m
        s_ok = s == exp_s
        if not (m_ok and s_ok):
            ok = False
        print(f"  {'✅' if m_ok and s_ok else '❌'} {code!r:>10} → market={m!r}({exp_m!r}) symbol={s!r}({exp_s!r})")

    print("\n=== is_trading_now 测试 ===")
    from datetime import datetime as _dt
    # 2026-08-14 是周五，10:00 在 A 股/港股早盘时段内
    t1 = _dt(2026, 8, 14, 10, 0)
    # 11:45 在港股早盘(9:30-12:00)，但 A 股午休
    t2 = _dt(2026, 8, 14, 11, 45)
    # 12:30 港股午休
    t3 = _dt(2026, 8, 14, 12, 30)
    # 周六
    t4 = _dt(2026, 8, 15, 10, 0)
    for market, dt, exp in [
        ('a', t1, True), ('hk', t1, True),
        ('a', t2, False), ('hk', t2, True),
        ('hk', t3, False),
        ('a', t4, False),
    ]:
        got = is_trading_now(market, dt)
        status = '✅' if got == exp else '❌'
        if got != exp:
            ok = False
        print(f"  {status} is_trading_now({market!r}, {dt.strftime('%F %H:%M')}) = {got} (expected {exp})")

    print(f"\n{'全部通过 ✅' if ok else '有失败 ❌'}")