tradingSystem/app/services/market.py

# -*- coding: utf-8 -*-
"""
行情与参考位取数 (best-effort, 取不到一律返回 None 由上层降级)
==============================================================
现价: Redis db13 (SIGNAL_REDIS_DB_QUOTES), key `tushare:rt_min:1MIN:{600000.SH}`,
      值为当日分钟 K 线数组 —— 与 bionic_trader 既有读法完全一致, 不另立口径。
参考位: 主口径取决策系统 strategy_daily_results 的支撑/压力; 日龄超 PMS_REF_STALE_TDAYS
      转兜底自算 (设计 §13): 支撑 = max(MA20, 近20日低点×1.01), 压力 = 近60日高点,
      止损 = 底仓成本 − N×ATR 与支撑取高。因子分表 gp_stock_factor_pro_YYYYMM 经 153 代理
      逐表单查 (跨分表 UNION 会被代理拒绝, 沿用 bionic 的踩坑结论)。
"""
from __future__ import annotations

import json
import logging
import threading
from datetime import datetime, timedelta

from config.settings import settings
from app.db.session import fetch_all
from app.repo import downstream_repo
from app.services import param_store

logger = logging.getLogger("pms.market")

_redis = None
_lock = threading.Lock()
QUOTE_KEY = "tushare:rt_min:1MIN:{code}"


def _r():
    """行情 Redis 客户端。

    强制 RESP2 (protocol=2): 目标 Redis 是 6.0 以前的版本, 不认 RESP3 的 HELLO 命令,
    而 redis-py 6.x 默认按 RESP3 握手, 会直接报 "unknown command `HELLO`" 全盘取不到价。
    requirements.txt 已把 redis 锁在 5.x, 这里再显式声明一道 (老版本 redis-py 不认这个
    参数, 用 TypeError 兜回去)。
    """
    global _redis
    if _redis is not None:
        return _redis
    with _lock:
        if _redis is None:
            import redis
            kw = dict(host=settings.SIGNAL_REDIS_HOST, port=settings.SIGNAL_REDIS_PORT,
                      password=settings.SIGNAL_REDIS_PASSWORD or None,
                      db=settings.SIGNAL_REDIS_DB_QUOTES, decode_responses=True,
                      socket_timeout=settings.SIGNAL_REDIS_SOCKET_TIMEOUT)
            try:
                _redis = redis.Redis(protocol=2, **kw)
            except TypeError:
                _redis = redis.Redis(**kw)
    return _redis


def _bars(code: str):
    try:
        raw = _r().get(QUOTE_KEY.format(code=code))
        if not raw:
            return None
        bars = json.loads(raw)
        return bars if isinstance(bars, list) and bars else None
    except Exception as e:
        logger.warning("实时行情读取失败 [%s]: %s", code, e)
        return None


def get_price(ts_code: str):
    bars = _bars(ts_code)
    if not bars:
        return None
    try:
        px = float(bars[-1].get("close") or 0)
        return px if px > 0 else None
    except (TypeError, ValueError):
        return None


def get_prices(codes) -> dict:
    """批量取现价 (逐 key GET; 数量级为持仓数, 无需 pipeline 复杂化)。"""
    out = {}
    for c in codes or []:
        out[c] = get_price(c)
    return out


def day_snapshot(ts_code: str) -> dict:
    """当日分钟线聚合: 现价/开盘/最高/最低/均价(VWAP近似)/涨幅 —— 择时与不追高检查用。"""
    bars = _bars(ts_code)
    if not bars:
        return {}
    try:
        pairs = [(float(b.get("close") or 0), float(b.get("vol") or 0)) for b in bars]
        pairs = [(c, v) for c, v in pairs if c > 0]     # 坏 bar 价量成对剔除, 不错位 (2026-08-28)
        closes = [c for c, _ in pairs]
        if not closes:
            return {}
        highs = [float(b.get("high") or b.get("close") or 0) for b in bars]
        lows = [float(b.get("low") or b.get("close") or 0) for b in bars if
                float(b.get("low") or b.get("close") or 0) > 0]
        opens = float(bars[0].get("open") or closes[0])
        vols = [v for _, v in pairs]
        amt = sum(c * v for c, v in zip(closes, vols))
        vwap = (amt / sum(vols)) if sum(vols) > 0 else sum(closes) / len(closes)
        px = closes[-1]
        return {"price": px, "open": opens, "high": max(highs) if highs else px,
                "low": min(lows) if lows else px, "vwap": round(vwap, 3),
                "day_chg_from_open": (px / opens - 1) if opens else None,
                "bar_time": bars[-1].get("time"), "bars": len(bars)}
    except (TypeError, ValueError) as e:
        logger.warning("行情聚合失败 [%s]: %s", ts_code, e)
        return {}


# ================================================================ 参考位
def _shard_tables(months: int = 4) -> list:
    now = datetime.now()
    out, d = [], now
    for _ in range(max(1, months)):
        out.append(f"gp_stock_factor_pro_{d.strftime('%Y%m')}")
        d = (d.replace(day=1) - timedelta(days=1))
    return out


def _factor_rows(ts_code: str, days: int = 70) -> list:
    """逐分表单查 (代理要求单表), 合并后取最近 days 条。symbol 先点式后前缀式试。

    回看窗至少 15 个自然日 (2026-08-28 审查修): 原来按 days×1.6 算, days=3 时只回看
    4.8 天 —— 春节国庆停市 7~9 个自然日, 长假后第一个交易日最近一根日线落在窗外,
    昨收取空、候选池整日空掉 (正是 2026-07-31 那次故障在长假后的复发形态)。"""
    since = (datetime.now() - timedelta(days=max(15, int(days * 2.5)))).strftime("%Y-%m-%d")
    rows = []
    for tbl in _shard_tables():
        for sym in (ts_code, downstream_repo.to_prefix(ts_code)):
            try:
                part = fetch_all(
                    f"SELECT trade_date, close_qfq, high_qfq, low_qfq, atr_qfq FROM {tbl} "
                    f"WHERE symbol = :sym AND trade_date >= :since", {"sym": sym, "since": since})
            except Exception:
                part = []
            if part:
                rows.extend(part)
                break
    rows = [r for r in rows if r.get("close_qfq")]
    rows.sort(key=lambda r: str(r.get("trade_date")))
    return rows[-days:]


def self_calc_refs(ts_code: str, base_cost=None) -> dict:
    """兜底自算参考位 (设计 §13)。数据不足返回 {}。"""
    rows = _factor_rows(ts_code)
    if len(rows) < 20:
        return {}
    close = [float(r["close_qfq"]) for r in rows]
    low = [float(r.get("low_qfq") or r["close_qfq"]) for r in rows]
    high = [float(r.get("high_qfq") or r["close_qfq"]) for r in rows]
    atr = None
    for r in reversed(rows):
        if r.get("atr_qfq"):
            atr = float(r["atr_qfq"])
            break
    ma20 = sum(close[-20:]) / 20
    support = max(ma20, min(low[-20:]) * 1.01)
    pressure = max(high[-60:]) if len(high) >= 60 else max(high)
    stop = None
    if base_cost and atr:
        stop = max(float(base_cost) - param_store.get_float("PMS_STOP_ATR_MULT", 2.0) * atr,
                   support)
    elif atr:
        stop = max(close[-1] - param_store.get_float("PMS_STOP_ATR_MULT", 2.0) * atr, support)
    return {"support": round(support, 3), "pressure": round(pressure, 3),
            "stop": round(stop, 3) if stop else None, "atr": atr, "ma20": round(ma20, 3),
            "source": "self_calc", "bars": len(rows)}


_ma_cache = {"day": None, "data": {}}


def get_last_close(ts_code: str):
    """上一交易日收盘 (日线 close_qfq)。**盘前定价的兜底** —— 按日缓存。

    db13 里是**当日**分钟线, 盘前那张 key 根本不存在, get_price 必然返回 None。
    2026-07-31 盘前实测: 上游计划前 30 只全部"无价", 候选池会整个空掉。计划价只用来
    算批次数量 (真正的委托价由择时环节按实时行情现定), 用昨收完全够用。
    """
    today = datetime.now().strftime("%Y%m%d")
    key = f"PC:{ts_code}"
    if _ma_cache["day"] != today:
        _ma_cache.update({"day": today, "data": {}})
    if key in _ma_cache["data"]:
        return _ma_cache["data"][key]
    val, failed = None, False
    try:
        rows = _factor_rows(ts_code, days=3)
        closes = [float(r["close_qfq"]) for r in rows if r.get("close_qfq")]
        if closes:
            val = round(closes[-1], 3)
    except Exception as e:
        failed = True          # 异常不落当日缓存: 盘前一次库抖不该让该票全天无昨收 (2026-08-28)
        logger.warning("昨收取数失败 [%s] (本次不缓存, 下次重查): %s", ts_code, e)
    if not failed:
        _ma_cache["data"][key] = val
    return val


def day_volume(ts_code: str) -> dict:
    """当日累计成交量与它实际覆盖的交易时段 (2026-09-09, 解锁重问的量能确认要用)。

    返回 {"vol": 股, "frac": 覆盖占全日的比例, "from": "10:11", "to": "15:00", "minutes": 199}
    取不到返回 {}。

    **为什么要带 frac 出来, 而不是让调用方按当前时刻算**: 实测 09-09 15:00,
    601126.SH 的分钟线首根是 10:11 不是 09:30 —— db13 里这张 key 是滚动窗口, 不保证
    从开盘存起。按当前时刻算已过时段, 分母会把没存进来的那段也算上, 量比被系统性压低,
    「量能确认」这道条件于是永远过不了。所以折算的分母只能是**这份数据真正覆盖的时段**。

    单位: 分钟线的 vol 是**股**, 因子表的日线 vol 是**手**。两边差 100 倍, 换算在
    reask_rules.vol_ratio 里做, 那里有单测钉着。
    """
    bars = _bars(ts_code)
    if not bars:
        return {}
    try:
        vol = sum(float(b.get("vol") or 0) for b in bars)
        t0, t1 = str(bars[0].get("time") or ""), str(bars[-1].get("time") or "")
        m0, m1 = _hhmm_min(t0), _hhmm_min(t1)
        if m0 is None or m1 is None or m1 < m0:
            return {}
        mins = _trading_minutes_between(m0, m1)
        if mins <= 0:
            return {}
        return {"vol": vol, "minutes": mins, "frac": round(mins / 240.0, 4),
                "from": t0[11:16], "to": t1[11:16], "bars": len(bars)}
    except (TypeError, ValueError) as e:
        logger.warning("当日成交量聚合失败 [%s]: %s", ts_code, e)
        return {}


def _hhmm_min(t: str):
    """"2026-09-09 10:11:00" → 611 (当天第几分钟)。认不出回 None。"""
    try:
        hh, mm = t[11:13], t[14:16]
        return int(hh) * 60 + int(mm)
    except (ValueError, IndexError):
        return None


def _trading_minutes_between(m0: int, m1: int) -> int:
    """两个时刻之间的**交易**分钟数 (跨午休不算休市那 90 分钟)。"""
    AM0, AM1, PM0, PM1 = 9 * 60 + 30, 11 * 60 + 30, 13 * 60, 15 * 60
    lo, hi = max(m0, AM0), min(m1, PM1)
    if hi <= lo:
        return 0
    n = 0
    for a, b in ((AM0, AM1), (PM0, PM1)):
        n += max(0, min(hi, b) - max(lo, a))
    return n


def get_vol_base5(ts_code: str):
    """近五个交易日的日均成交量 (单位: **手**, 因子表口径)。按日缓存, 取不到回 None。

    单独一条 SQL 而不是扩 _factor_rows 的 SELECT: 那个函数在候选池装配的热路径上被几十只
    票各调一次, 给它加一列会让每一次都多搬一份用不上的数据。这条只在解锁判定里按票调一次。
    """
    today = datetime.now().strftime("%Y%m%d")
    key = f"VB5:{ts_code}"
    if _ma_cache["day"] != today:
        _ma_cache.update({"day": today, "data": {}})
    if key in _ma_cache["data"]:
        return _ma_cache["data"][key]
    val, failed = None, False
    since = (datetime.now() - timedelta(days=30)).strftime("%Y-%m-%d")
    try:
        rows = []
        for tbl in _shard_tables():
            # 因子表的 symbol 实测是前缀式 (SH601126), 但两种写法都试 —— 与 _factor_rows 同手法,
            # 上游哪天改了写法这里不至于静默取空。
            for sym in (downstream_repo.to_prefix(ts_code), ts_code):
                try:
                    part = fetch_all(
                        f"SELECT trade_date, vol FROM {tbl} "
                        f"WHERE symbol = :sym AND trade_date >= :since", {"sym": sym, "since": since})
                except Exception:
                    part = []
                if part:
                    rows.extend(part)
                    break
        rows = [r for r in rows if r.get("vol")]
        rows.sort(key=lambda r: str(r.get("trade_date")))
        vols = [float(r["vol"]) for r in rows[-5:]]
        if len(vols) >= 3:              # 三根就够算日均; 少于三根宁可不给
            val = round(sum(vols) / len(vols), 2)
    except Exception as e:
        failed = True
        logger.warning("五日均量取数失败 [%s] (本次不缓存): %s", ts_code, e)
    if not failed:
        _ma_cache["data"][key] = val
    return val


def plan_price(ts_code: str) -> dict:
    """规划用价: 实时价优先, 盘前/停更回落昨收。返回 {price, source}。

    source: realtime / prev_close / none —— **一定要带出来**。拿昨收当现价去做"不追高"
    这类判断会出错, 所以只给规划期定量用, 判断类的检查仍走 day_snapshot。
    """
    px = get_price(ts_code)
    if px:
        return {"price": px, "source": "realtime"}
    px = get_last_close(ts_code)
    if px:
        return {"price": px, "source": "prev_close"}
    return {"price": None, "source": "none"}


def get_ma5(ts_code: str):
    """MA5 (规则闸「不追高」用)。按日缓存 —— 因子分表是日频数据, 盘中重复查没意义。"""
    today = datetime.now().strftime("%Y%m%d")
    if _ma_cache["day"] != today:
        _ma_cache.update({"day": today, "data": {}})
    if ts_code in _ma_cache["data"]:
        return _ma_cache["data"][ts_code]
    val, failed = None, False
    try:
        rows = _factor_rows(ts_code, days=10)
        closes = [float(r["close_qfq"]) for r in rows if r.get("close_qfq")]
        if len(closes) >= 5:
            val = round(sum(closes[-5:]) / 5, 3)
    except Exception as e:
        failed = True
        logger.warning("MA5 取数失败 [%s] (本次不缓存, 下次重查): %s", ts_code, e)
    if not failed:
        _ma_cache["data"][ts_code] = val
    return val


def get_high5(ts_code: str):
    """近 5 个交易日最高价 (动作引擎「创 5 日新高」用)。同样按日缓存。"""
    today = datetime.now().strftime("%Y%m%d")
    key = f"H5:{ts_code}"
    if _ma_cache["day"] != today:
        _ma_cache.update({"day": today, "data": {}})
    if key in _ma_cache["data"]:
        return _ma_cache["data"][key]
    val, failed = None, False
    try:
        rows = _factor_rows(ts_code, days=10)
        highs = [float(r.get("high_qfq") or r.get("close_qfq") or 0) for r in rows]
        highs = [h for h in highs if h > 0]
        if len(highs) >= 5:
            val = round(max(highs[-5:]), 3)
    except Exception as e:
        failed = True
        logger.warning("5日高点取数失败 [%s] (本次不缓存, 下次重查): %s", ts_code, e)
    if not failed:
        _ma_cache["data"][key] = val
    return val


def get_refs(ts_code: str, *, base_cost=None) -> dict:
    """参考位: 决策系统主口径 → 日龄超期/缺失时兜底自算 → 都拿不到返回 source=none。"""
    stale_days = param_store.get_int("PMS_REF_STALE_TDAYS", 3)
    try:
        r = downstream_repo.fetch_refs(ts_code)
    except Exception as e:
        logger.warning("读决策系统结论失败 [%s]: %s", ts_code, e)
        r = None
    if r and (r.get("support") or r.get("pressure")):
        age = _age_days(r.get("trade_date"))
        if age is not None and age <= stale_days * 2:      # 自然日宽松换算交易日
            return {"support": r.get("support"), "pressure": r.get("pressure"),
                    "stop": r.get("support"), "source": "bionic",
                    "trade_date": str(r.get("trade_date")), "age_days": age}
    try:
        s = self_calc_refs(ts_code, base_cost=base_cost)
    except Exception as e:
        logger.warning("自算参考位失败 [%s]: %s", ts_code, e)
        s = {}
    if s:
        s["note"] = "兜底口径 (决策系统结论缺失或停更)"
        return s
    return {"support": None, "pressure": None, "stop": None, "source": "none",
            "note": "参考位不可用 —— 敞口无法计算, 相关动作按保守处理"}


def _age_days(trade_date):
    if not trade_date:
        return None
    try:
        s = str(trade_date)[:10].replace("-", "")
        d = datetime.strptime(s, "%Y%m%d").date()
        return (datetime.now().date() - d).days
    except (ValueError, TypeError):
        return None