160 lines
5.6 KiB
Python
160 lines
5.6 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""market_config.py — 市场抽象层（阶段1：港股接入核心结构件）
|
||
|
||
市场判断 / 行情符号 / 市场配置的**唯一事实源**，收敛散落在
|
||
market_data / price_monitor / stale_push_wlin / 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 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/stale_push_wlin/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
|
||
|
||
|
||
# ── 模块自检 ────────────────────────────────────────────────────────────
|
||
|
||
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 '有失败 ❌'}")
|