tradingSystem/app/services/param_store.py

342 lines
18 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

# -*- coding: utf-8 -*-
"""
运行参数中心 (页面调参即时生效的唯一入口)
==========================================
纪律 (config/settings.py 头部第 3 条): 业务代码**禁止**直接读 settings 取业务参数,
一律走 ParamStore —— 表值 (pms_runtime_param) 优先于文件初值, 页面改完立刻生效。
除 settings 里声明的业务参数外, 另有一批「运行态开关」也落同一张表 (DDL 无需新增表):
PMS_GLOBAL_BUY_HALT / PMS_GLOBAL_EXEC_HALT 全局暂停买入 / 暂停执行
PMS_BRAKE_UNTIL 组合刹车解除日 (YYYYMMDD, 0=未刹车)
PMS_HIGH_WATER 组合高水位 (刹车判定基准)
PMS_REPLAY_CURSOR 成交回放游标
PMS_RECON_STREAK 连续对账不一致天数
PMS_SECTOR_CAP_* 单行业上限 (SECTOR_CAP 命令写入)
"""
from __future__ import annotations
import logging
import threading
import time
from config.settings import settings
from app.repo import pms_repo
logger = logging.getLogger("pms.params")
CACHE_TTL = 5.0 # 秒; 页面改参后最迟 5 秒全进程可见 (worker 多进程各持一份)
# 不允许页面修改的基础设施键 (连接串等只在 .env 维护)
INFRA_PREFIX = ("PROXY_DB", "SOURCE_DB", "DB_MYSQL", "SIGNAL_REDIS", "PMS_REDIS", "PMS_WEB")
# 密钥: 既不可改, 也**不可读** —— 连页面快照里都不出现。
# 协议 QMT_WS_PROTOCOL.md §10.1.1 明确要求「只走 .env, 不入库、不进 ParamStore、
# 不写进代码」。它们同样以 PMS_ 开头, 不单列的话会被 _editable_keys() 当成普通业务参数:
# 页面能改 (于是私钥 seed 落进 pms_runtime_param 表), snapshot() 还会把它原文显示出来。
SECRET_KEYS = ("PMS_QMT_SIGN_SEED_HEX", "PMS_QMT_PEER_PUBKEY_B64")
# 运行态开关: key -> (默认值, 类型, 说明)
RUNTIME_EXTRA = {
"PMS_GLOBAL_BUY_HALT": (False, bool, "全局暂停买入 (HALT_BUY 命令置位)"),
"PMS_GLOBAL_EXEC_HALT": (False, bool, "全局暂停执行 / 休假模式"),
"PMS_BRAKE_UNTIL": (0, int, "组合刹车解除日 YYYYMMDD, 0=未刹车"),
"PMS_HIGH_WATER": (0.0, float, "组合市值高水位 (刹车判定基准)"),
"PMS_REPLAY_CURSOR": ("", str, "成交回放游标 (trading_order.order_id)。留空=首次启动时自动对齐到当前最新成交且不追认历史; 填某个 order_id=从它之后开始; 填 ALL=从头全量回放"),
"PMS_RECON_STREAK": (0, int, "连续对账不一致天数"),
}
# 页面展示用的中文说明 (settings.py 用行尾注释, pydantic 取不到, 故在此集中维护)
DESC = {
"PMS_TOTAL_SCALE": "总操作规模 (元) —— 所有百分比约束的分母",
"PMS_PORTFOLIO_CAP": "总仓上限 (占规模)", "PMS_STOCK_CAP": "单股上限 (占规模)",
"PMS_STOCK_TARGET_DEFAULT": "默认单股目标仓位", "PMS_MAX_NAMES": "最大持仓只数",
"PMS_CASH_RESERVE": "预留现金比例 (永不动用, 与总仓上限双重约束)",
"PMS_AUTONOMY": "自主档位 full / propose_only / off",
"PMS_BATCH_SPLIT": "单股分批比例 (底仓/回踩补足/盈利加仓)",
"PMS_CUSHION_SOLID": "安全垫厚垫线 (解锁盈利加仓)",
"PMS_TRIM_PEAK": "保垫减仓: 垫子峰值门槛", "PMS_TRIM_GIVEBACK": "保垫减仓: 回吐比例门槛",
"PMS_DCA_TRIGGERS": "补仓评估档 (浮亏)", "PMS_DCA_DEEP_CONFIRM": "此档及更深永远需用户确认",
"PMS_DCA_MAX_RATIO": "补仓上限 (占底仓)", "PMS_NO_CHASE_MA5": "距 MA5 超此幅度不追买",
"PMS_BUILD_WINDOW_TDAYS": "建仓期窗口 (交易日)",
"PMS_FILL_MAX_LOSS": "浮亏深于此不走回踩补足",
"PMS_WEAK_NEG_DAYS": "降仓清弱票: 安全垫连续为负天数",
"PMS_PROPOSAL_TTL_HOURS": "自主提议待确认有效期 (小时)",
"PMS_CANDIDATE_SOURCE": "升仓候选池来源: plan_api=上游选股计划接口 / buy_plan=旧 trading_buy_plan 表 / both=并集",
"PMS_PLAN_API_BASE": "上游选股计划接口根地址; 留空=停用, 此时候选池恒为空并告警",
"PMS_PLAN_API_PATH": "计划接口路径 (默认 /plan)",
"PMS_PLAN_TIMEOUT": "计划接口超时 (秒)",
"PMS_PLAN_CACHE_SEC": "计划缓存秒数 (上游日频产出)",
"PMS_PLAN_TOP": "发给上游的 top: 主榜要多少条 (上游默认 20; 0=不传)",
"PMS_PLAN_OBS_TOP": "发给上游的 obs_top: 观察档要多少条 (上游默认 10; 0=不传)",
"PMS_PLAN_THEME_CAP": "发给上游的 theme_cap: 上游侧同主题限额 (上游默认 5; 设大值=让上游别裁, 由 PMS 自己裁; 0=不传)",
"PMS_PLAN_THEME_CAP_LOCAL": "PMS 侧同主题限额, 在 TOP_N 截断之前按 score 序生效, 0=不限。防止宽池子里前 N 名被单一主题垄断、切完再被规则闸拦掉",
"PMS_PLAN_TOP_N": "主榜按 score 降序取前 N 只进候选池",
"PMS_PLAN_TIERS": "传导档白名单 (如 强传导), 留空=不按档过滤",
"PMS_PLAN_INCLUDE_OBSERVE": "观察档是否进候选池",
"PMS_PLAN_MIN_SCORE": "候选 score 下限, 0=不设",
"PMS_PLAN_MIN_SOURCES": "候选 evidence.n_sources 下限, 0=不设",
"PMS_PLAN_MIN_UPSIDE": "候选预期空间下限 (相对现价, 0.5=+50%), 0=不设。只过滤不参与排序; 设了就会把 upside 缺失的行(含整个观察档)一起挡掉",
"PMS_PLAN_STALE_TDAYS": "计划日龄超此交易日数即判过期拒用 (防上游停更时拿旧榜当今天)",
"PMS_PLAN_EXCLUDE_ST": "剔除 ST/*ST/退市整理 (上游名单明确不剔除, 这道闸在 PMS 侧)",
"PMS_PLAN_THEME_SYNC": "把 theme 灌进 pms_industry_map。默认关 —— theme 是事件驱动的传导主题(覆盖约三成、词表不规范), 当行业标签会让集中度约束天天漂; 行业源请用 gp_stock_category",
"PMS_PLAN_QUERY_EXTRA": "计划接口附加查询串逃生口 (上游加了新参数时不用改代码); top/obs_top/theme_cap 请用各自的显式参数, 同名键以显式参数为准",
"PMS_SECTOR_SOURCE": "行业划分数据源: 空=约束停用 / custom_table (推荐, 由上游计划的 theme 灌数) / gp_stock_category",
"PMS_SECTOR_MAX_NAMES": "同行业最大持仓只数 (硬拦截)",
"PMS_SECTOR_MAX_RATIO": "同行业最大占总仓比例 (硬拦截)",
"PMS_EXEC_WINDOW_TDAYS": "任务命令默认执行窗口 (交易日)",
"PMS_SELL_AVOID_OPEN_MIN": "卖出避开开盘 N 分钟",
"PMS_BUY_HALT_DAYUP": "当日涨幅超此停止买入 (不追高)",
"PMS_EOD_FORCE_TIME": "当日配额兜底时点", "PMS_EOD_FORCE_DISCOUNT": "兜底限价系数 (卖出)",
"PMS_MIN_LOT_MERGE": "一手检查: 批次自动合并",
"PMS_DISPATCH_EXPIRE_MIN": "指令下发后未被接受的过期时间 (分钟)",
"PMS_DISPATCH_MODE": "下发通道: shadow=只记账待人工 / ws=WebSocket 直连 QMT (需先起 pms-ws)",
"PMS_EXEC_SLICES": "当日配额分几笔出手",
"PMS_ORDER_TTL_MIN": "单笔挂单有效期 (交易分钟), 到点下游自动撤; 兜底单不受限",
"PMS_RISK_WARN_ENTRY": "单笔敞口告警线 (占规模)", "PMS_RISK_WARN_PORTFOLIO": "组合敞口告警线",
"PMS_BRAKE_DRAWDOWN": "组合刹车: 自高水位回撤", "PMS_BRAKE_DAYS": "刹车持续交易日",
"PMS_STOP_ATR_MULT": "自算止损参考: 成本 N×ATR",
"PMS_REF_STALE_TDAYS": "决策系统结论日龄超此转自算兜底",
"PMS_JUDGE_ENABLED": "研判闸开关", "PMS_JUDGE_ACTIONS": "需过研判闸的动作",
"PMS_JUDGE_TIMEOUT": "研判超时 (秒) → 降级 propose_only",
"PMS_JUDGE_API_BASE": "决策系统 PMS 研判接口根地址; 留空=未接通, 自动降级人工确认",
"PMS_JUDGE_PATH": "研判接口路径 (bionic 侧配套改造后确定)",
"PMS_T0_RATIO_MAX": "T 仓硬上限 (占持仓)", "PMS_T0_PULLBACK_PCT": "正T: 距当日高点回落触发",
"PMS_T0_RALLY_PCT": "反T: 日内涨幅触发", "PMS_T0_ROUND_TARGET": "单次T目标价差",
"PMS_T0_CLOSE_TIME": "T仓强制平回时点", "PMS_T0_STOCK_DAY_LOSS": "单票当日T亏熔断",
"PMS_T0_GLOBAL_DAY_LOSS": "全局当日T亏熔断",
"PMS_REPLAY_INTERVAL_MIN": "成交回放间隔 (分钟)", "PMS_RECON_ALARM_DAYS": "连续不一致升级天数",
"PMS_SIGNAL_ENABLED": "是否消化决策系统盘中信号",
"PMS_SIGNAL_GROUP": "信号消费组名 (独立于 trading_service, 互不抢消息)",
"PMS_SIGNAL_SELL_CONF_MIN": "卖出信号消化门槛 (低于此不动)",
"PMS_SIGNAL_AUTO_EXIT_CONF": "卖出信号直接清仓门槛 (之间则落提议)",
"PMS_SIGNAL_TRIM_RATIO": "中等置信度卖出信号的减仓比例",
"PMS_QMT_WS_URL": "QMT 执行服务 WebSocket 端点",
"PMS_QMT_WS_ENABLED": "pms-ws 常驻进程总开关 (关=空转不连接, 此时一律拒发)",
"PMS_QMT_ACK_BATCH": "累积确认: 每落库 N 条发一次 ack_seq",
"PMS_QMT_ACK_INTERVAL_SEC": "累积确认: 或每 N 秒发一次 (与条数取先到)",
"PMS_QMT_HEARTBEAT_SEC": "协议 ping 间隔 (秒)",
"PMS_QMT_IDLE_TIMEOUT_SEC": "超过此秒数未收到对端任何消息即断开重连",
"PMS_QMT_OUTBOX_POLL_SEC": "出口队列轮询间隔 (秒)",
"PMS_QMT_HEARTBEAT_DB_SEC": "ws 进程写存活心跳的间隔 (秒)",
"PMS_QMT_HEARTBEAT_STALE_SEC": "心跳陈旧超此秒数 → 判定 ws 进程已死, 指令一律拒发",
"PMS_QMT_CONNECT_TIMEOUT_SEC": "建连超时 (秒)",
"PMS_QMT_SEND_MAX_ATTEMPTS": "单张委托发送重试上限, 试满置 SEND_FAILED 等人工",
}
# loaded 标记必不可少: 不能用「data 是否为空」判断缓存是否有效 ——
# 参数表为空 (还没在页面改过参数) 或读表失败时 data 都是 {}, 那样每次 get() 都会穿透去连库,
# 健康时是白白打表, 故障时是刷屏重试。实测 check_db 一次跑出 50 条重复告警即此故。
_cache = {"at": 0.0, "data": {}, "error": None, "loaded": False}
_lock = threading.Lock()
def _editable_keys() -> dict:
"""可调业务参数 = settings 中 PMS_ 开头、非基础设施、且非密钥的字段。"""
out = {}
for name, field in type(settings).model_fields.items():
if not name.startswith("PMS_"):
continue
if any(name.startswith(p) for p in INFRA_PREFIX) or name in SECRET_KEYS:
continue
out[name] = field
return out
def _coerce(value, target_type):
if target_type is bool:
if isinstance(value, bool):
return value
return str(value).strip().lower() in ("1", "true", "yes", "on", "y")
if target_type is int:
return int(float(value))
if target_type is float:
return float(value)
return str(value)
def _type_of(key: str):
f = _editable_keys().get(key)
if f is not None:
return f.annotation
if key in RUNTIME_EXTRA:
return RUNTIME_EXTRA[key][1]
return str
def refresh(force: bool = False) -> dict:
"""拉一次表值 (带 TTL 缓存)。连库失败不抛异常 —— 退回文件初值并记 error。"""
now = time.time()
if not force and _cache["loaded"] and now - _cache["at"] < CACHE_TTL:
return _cache["data"]
with _lock:
try:
rows = pms_repo.all_params()
_cache.update({"data": rows, "at": now, "error": None, "loaded": True})
except Exception as e:
msg = f"{type(e).__name__}: {e}"
changed = _cache.get("error") != msg
# 失败也要更新时间戳并置 loaded, 否则下一次 get() 立刻又去连库
_cache.update({"at": now, "error": msg, "loaded": True})
if changed: # 同一个错误只吼一次, 恢复或换错才再吼
logger.warning("参数表读取失败, 退回 settings 初值: %s", e)
return _cache["data"]
def get(key: str, default=None):
"""取参数当前值: 表值优先 → settings 初值 → RUNTIME_EXTRA 默认 → default。
密钥一律返回空串 —— 想拿签名密钥只有一条路: 直接读 settings (即 .env)。
堵死这里是为了让「密钥不进 ParamStore」这句话在代码里成立, 而不只是写在文档上。
"""
if key in SECRET_KEYS:
return ""
rows = refresh()
t = _type_of(key)
if key in rows:
try:
return _coerce(rows[key]["param_value"], t)
except (TypeError, ValueError):
logger.warning("参数 %s 表值非法 (%s), 退回初值", key, rows[key]["param_value"])
if hasattr(settings, key):
return getattr(settings, key)
if key in RUNTIME_EXTRA:
return RUNTIME_EXTRA[key][0]
return default
def get_float(key, default=0.0):
try:
return float(get(key, default))
except (TypeError, ValueError):
return default
def get_int(key, default=0):
try:
return int(float(get(key, default)))
except (TypeError, ValueError):
return default
def get_bool(key, default=False):
v = get(key, default)
return v if isinstance(v, bool) else str(v).strip().lower() in ("1", "true", "yes", "on")
def get_list(key, default=None, sep=","):
v = get(key, None)
if v in (None, ""):
return list(default or [])
return [x.strip() for x in str(v).split(sep) if x.strip()]
def get_tuple_floats(key, default=(0.5, 0.25, 0.25)):
try:
vals = tuple(float(x) for x in str(get(key, "")).split(",") if str(x).strip())
return vals or tuple(default)
except (TypeError, ValueError):
return tuple(default)
def set_param(key: str, value, updated_by: str = "user") -> dict:
"""页面改参入口。校验键名与类型后落表, 并立即失效缓存。"""
editable = _editable_keys()
if key not in editable and key not in RUNTIME_EXTRA and not key.startswith("PMS_SECTOR_CAP_"):
return {"ok": False, "error": f"参数 {key} 不可修改 (基础设施参数只在 .env 维护)"}
t = _type_of(key)
try:
v = _coerce(value, t)
except (TypeError, ValueError):
return {"ok": False, "error": f"参数 {key} 类型应为 {getattr(t, '__name__', t)}, "
f"收到 {value!r}"}
bad = _range_check(key, v)
if bad:
return {"ok": False, "error": bad}
try:
pms_repo.set_param(key, v, updated_by)
except Exception as e:
return {"ok": False, "error": f"写入失败: {type(e).__name__}: {e}"}
_cache["loaded"] = False
refresh(force=True)
return {"ok": True, "key": key, "value": v}
_RANGES = {
"PMS_PORTFOLIO_CAP": (0, 1), "PMS_STOCK_CAP": (0, 1), "PMS_STOCK_TARGET_DEFAULT": (0, 1),
"PMS_CASH_RESERVE": (0, 1), "PMS_CUSHION_SOLID": (0, 1), "PMS_TRIM_GIVEBACK": (0, 1),
"PMS_DCA_MAX_RATIO": (0, 1), "PMS_SECTOR_MAX_RATIO": (0, 1), "PMS_T0_RATIO_MAX": (0, 0.3334),
"PMS_MAX_NAMES": (1, 200), "PMS_TOTAL_SCALE": (0, 10 ** 12),
"PMS_EXEC_WINDOW_TDAYS": (1, 20), "PMS_BRAKE_DAYS": (0, 30),
"PMS_PLAN_TOP_N": (1, 1000), "PMS_PLAN_TIMEOUT": (1, 120),
"PMS_PLAN_CACHE_SEC": (5, 86400), "PMS_PLAN_STALE_TDAYS": (0, 20),
"PMS_PLAN_MIN_SOURCES": (0, 100), "PMS_PLAN_TOP": (0, 5000),
"PMS_PLAN_OBS_TOP": (0, 5000), "PMS_PLAN_THEME_CAP": (0, 5000),
"PMS_PLAN_THEME_CAP_LOCAL": (0, 1000),
}
def _range_check(key, v):
if key == "PMS_AUTONOMY" and v not in ("full", "propose_only", "off"):
return "PMS_AUTONOMY 只能是 full / propose_only / off"
if key == "PMS_SECTOR_SOURCE" and v not in ("", "custom_table", "gp_stock_category"):
return "PMS_SECTOR_SOURCE 只能是 空 / custom_table / gp_stock_category"
if key == "PMS_DISPATCH_MODE" and v not in ("shadow", "ws"):
return "PMS_DISPATCH_MODE 只能是 shadow / ws"
if key == "PMS_CANDIDATE_SOURCE" and v not in ("plan_api", "buy_plan", "both"):
return "PMS_CANDIDATE_SOURCE 只能是 plan_api / buy_plan / both"
lo_hi = _RANGES.get(key)
if lo_hi and isinstance(v, (int, float)) and not isinstance(v, bool):
lo, hi = lo_hi
if v < lo or v > hi:
return f"{key} 应在 [{lo}, {hi}] 区间, 收到 {v}"
return None
def snapshot() -> dict:
"""页面「参数设置」用: 每个可调参数的 当前值/来源/初值/说明。"""
rows = refresh()
out = {"params": [], "source_error": _cache.get("error")}
for key, field in sorted(_editable_keys().items()):
cur = get(key)
out["params"].append({
"key": key, "value": cur, "file_default": getattr(settings, key, None),
"source": "table" if key in rows else "file",
"type": getattr(field.annotation, "__name__", str(field.annotation)),
"desc": DESC.get(key, (field.description or "").strip()),
"updated_at": str(rows.get(key, {}).get("updated_at") or ""),
"updated_by": rows.get(key, {}).get("updated_by") or "",
})
for key, (dv, t, desc) in RUNTIME_EXTRA.items():
out["params"].append({
"key": key, "value": get(key), "file_default": dv,
"source": "table" if key in rows else "default",
"type": t.__name__, "desc": desc, "group": "runtime",
"updated_at": str(rows.get(key, {}).get("updated_at") or ""),
"updated_by": rows.get(key, {}).get("updated_by") or "",
})
return out
# ---------------------------------------------------------------- 常用组合读取
def sizing_params() -> dict:
"""方案生成器/规则闸共用的一组参数快照 (一次取齐, 避免逐项穿透缓存)。"""
return {
"scale": get_float("PMS_TOTAL_SCALE", 0),
"portfolio_cap": get_float("PMS_PORTFOLIO_CAP", 0.6),
"stock_cap": get_float("PMS_STOCK_CAP", 0.08),
"stock_target_default": get_float("PMS_STOCK_TARGET_DEFAULT", 0.06),
"max_names": get_int("PMS_MAX_NAMES", 15),
"cash_reserve": get_float("PMS_CASH_RESERVE", 0.0),
"autonomy": get("PMS_AUTONOMY", "propose_only"),
"batch_split": get_tuple_floats("PMS_BATCH_SPLIT"),
"min_lot_merge": get_bool("PMS_MIN_LOT_MERGE", True),
"sector_max_names": get_int("PMS_SECTOR_MAX_NAMES", 4),
"sector_max_ratio": get_float("PMS_SECTOR_MAX_RATIO", 0.40),
"sector_source": get("PMS_SECTOR_SOURCE", ""),
"exec_window_tdays": get_int("PMS_EXEC_WINDOW_TDAYS", 3),
"cushion_solid": get_float("PMS_CUSHION_SOLID", 0.03),
"weak_neg_days": get_int("PMS_WEAK_NEG_DAYS", 5),
"buy_halt": get_bool("PMS_GLOBAL_BUY_HALT", False),
"exec_halt": get_bool("PMS_GLOBAL_EXEC_HALT", False),
}