tradingSystem/app/repo/tech_repo.py

# -*- coding: utf-8 -*-
"""pms_tech_daily 单表访问 (2026-09-11 技术面接入)。

全市场每日技术面读数的落表与回看。**严格单表访问**: 每个函数只碰 pms_tech_daily 一张表。
批量落表走 execute_many + ON DUPLICATE KEY UPDATE (同一 (读数日, 代码) 重拉即覆盖);
回看按 (ts_code, data_date) 索引取近 N 行。保留窗口的门槛日由 tech_service 按交易日历算出,
再调 prune 删更旧的行。
"""
from __future__ import annotations

from datetime import datetime

from app.db.session import execute, execute_many, fetch_all, fetch_one

# 落表的业务列 (与 ddl_pms_v1.sql 的 pms_tech_daily 对齐; created_at/updated_at 由本层补北京时间)
_COLS = (
    "data_date", "ts_code", "stock_name",
    "boll_upper", "boll_mid", "boll_lower", "boll_bw_pct", "boll_pos", "boll_state", "boll_squeeze",
    "bbi", "bbi_upper", "bbi_lower", "bbi_pos", "bbi_state", "bbi_dist_pct",
    "sar_value", "sar_side", "sar_flip_days", "sar_dist_pct",
    "reanchored", "in_pool", "quality", "bars_used", "last_bar_date", "bars_lag", "algo_version",
)


def upsert_daily(rows: list) -> int:
    """批量落当日读数。rows 每项是 {列名: 值}, 至少含 data_date 与 ts_code。同 (日, 码) 覆盖。"""
    rows = [r for r in (rows or []) if r.get("data_date") and r.get("ts_code")]
    if not rows:
        return 0
    now = datetime.now()          # 北京时间 (容器时钟), 与项目写记录口径一致
    payload = []
    for r in rows:
        d = {c: r.get(c) for c in _COLS}
        d["ts"] = now
        payload.append(d)
    cols = ", ".join(_COLS) + ", created_at, updated_at"
    vals = ", ".join(f":{c}" for c in _COLS) + ", :ts, :ts"
    # ON DUPLICATE 子句用 VALUES(列) 引用插入值, 不再带绑定参数 (:col)。原因: pymysql 的
    # executemany 对 INSERT ... ON DUPLICATE 做多行合并优化, 只把 VALUES 子句的占位符按行
    # 展开, UPDATE 子句里的 :col (→ %s) 不展开却仍算参数, 5000 行批量时参数错位、SQL 里
    # 留下裸 % 报 1064 (2026-09-11 真机首验抓到; 脱库单测 mock 了落表测不到)。
    updates = ", ".join(f"{c} = VALUES({c})" for c in _COLS
                        if c not in ("data_date", "ts_code")) + ", updated_at = VALUES(updated_at)"
    return execute_many(
        f"INSERT INTO pms_tech_daily ({cols}) VALUES ({vals}) "
        f"ON DUPLICATE KEY UPDATE {updates}", payload)


def latest_date():
    """最新读数日 YYYYMMDD; 空表返回 None。"""
    r = fetch_one("SELECT MAX(data_date) AS d FROM pms_tech_daily")
    return int(r["d"]) if r and r.get("d") is not None else None


def count_on(data_date: int) -> int:
    """某读数日在库的行数 (状态与探活用)。"""
    r = fetch_one("SELECT COUNT(*) AS n FROM pms_tech_daily WHERE data_date = :d",
                  {"d": int(data_date)})
    return int(r["n"]) if r else 0


def distinct_dates(limit: int = 60) -> list:
    """最近 limit 个读数日, 降序。"""
    rows = fetch_all("SELECT DISTINCT data_date FROM pms_tech_daily "
                     "ORDER BY data_date DESC LIMIT :n", {"n": int(limit)})
    return [int(r["data_date"]) for r in rows]


def history(ts_code: str, *, since: int = 0, limit: int = 60) -> list:
    """一只票的近若干日读数, **升序** (最后一行最新)。since>0 时只取该日及以后。"""
    code = (ts_code or "").strip()
    if not code:
        return []
    p = {"c": code, "n": int(limit)}
    where = "ts_code = :c"
    if since:
        where += " AND data_date >= :since"
        p["since"] = int(since)
    rows = fetch_all(f"SELECT * FROM pms_tech_daily WHERE {where} "
                     f"ORDER BY data_date DESC LIMIT :n", p)
    return list(reversed(rows))


def history_multi(codes, *, since: int = 0) -> dict:
    """一批票各自的近日读数, 返回 {ts_code: [行, 升序]}。按 IN 取回再在内存分组。
    codes 多时分批 (每批 800) 防 SQL 过长; IN 占位符手动展开 (照 pms_repo.ledger_by_ref)。"""
    out: dict = {}
    uniq = [c for c in dict.fromkeys(str(x).strip() for x in (codes or []) if x) if c]
    if not uniq:
        return out
    for i in range(0, len(uniq), 800):
        chunk = uniq[i:i + 800]
        keys, p = [], {}
        for j, c in enumerate(chunk):
            keys.append(f":c{j}")
            p[f"c{j}"] = c
        where = f"ts_code IN ({', '.join(keys)})"
        if since:
            where += " AND data_date >= :since"
            p["since"] = int(since)
        rows = fetch_all(f"SELECT * FROM pms_tech_daily WHERE {where} "
                         f"ORDER BY ts_code ASC, data_date ASC", p)
        for r in rows:
            out.setdefault(r["ts_code"], []).append(r)
    return out


def prune(before_ymd: int) -> int:
    """删读数日早于 before_ymd 的行 (保留窗口门槛由调用方按交易日算)。返回删除行数。"""
    if not before_ymd:
        return 0
    return execute("DELETE FROM pms_tech_daily WHERE data_date < :d", {"d": int(before_ymd)})