# -*- 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)})