592 lines
27 KiB
Python
592 lines
27 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""上游信号只读快照 (2026-08-12: 只展示、不进任何下单/决策逻辑)。
|
|
|
|
把三系统盘中信号取最近若干条给页面看:
|
|
· 决策系统(bionic) 买卖广播 + 风控卖出动作 —— 208 db2 intraday_signals / db3 llm_sell_actions
|
|
· 盘中择时层(intraday_timing) BUY 入场 + 双向风控告警 —— 208 db2 intraday_signals / intraday_alerts
|
|
· mtf 资金异动 + 实况分关注/回避榜 —— 208 db2 stream:metrics / 214 db0 mr:board
|
|
|
|
**严格只读**: 一律 XREVRANGE / ZREVRANGE 取最近, 不建消费组、不 ACK、不写 —— 碰不到上游的消费与
|
|
投递, 也不和 PMS 现有信号消化(signal_service 的消费组)抢消息。连不通的源单独降级 (ok=False + error),
|
|
不影响其余与整页。取数带进程内缓存 (默认 8s), 不高频打上游。
|
|
`intraday_signals` 一条流两家都写、格式不同: 按 producer_id / entry_score 字段特征分成
|
|
「决策系统」与「择时层」两类展示。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import time
|
|
from datetime import datetime
|
|
|
|
from config.settings import settings
|
|
from app.services import param_store
|
|
|
|
logger = logging.getLogger("pms.upstream")
|
|
|
|
_clients = {}
|
|
_cache = {"at": 0.0, "data": None}
|
|
|
|
|
|
def _to() -> int:
|
|
return max(1, param_store.get_int("PMS_UPSTREAM_TIMEOUT_SEC", 3))
|
|
|
|
|
|
def _limit() -> int:
|
|
return max(1, param_store.get_int("PMS_UPSTREAM_LIMIT", 40))
|
|
|
|
|
|
def _alert_scan() -> int:
|
|
# 告警流扫更大窗口, 防高频源(如 price_notice)把风控告警挤出"最新N"。
|
|
return max(_limit(), param_store.get_int("PMS_UPSTREAM_ALERT_SCAN", 300))
|
|
|
|
|
|
def _alert_per_cat() -> int:
|
|
# 每个类型最多留最新 N 条, 高频源只占它自己那一档, 不挤别的类。
|
|
return max(1, param_store.get_int("PMS_UPSTREAM_ALERT_PER_CAT", 20))
|
|
|
|
|
|
def _alert_max_age_sec() -> int:
|
|
# 告警时间窗(秒): 0=不限龄。只用于给每条打"陈旧"标记(超龄置灰), **不删条目** ——
|
|
# 用户拍板"置灰标注"而非隐藏, 风控面板不丢信息。与每类上限正交: 先按龄标 stale, 再每类限量。
|
|
return max(0, param_store.get_int("PMS_UPSTREAM_ALERT_MAX_AGE_MIN", 60)) * 60
|
|
|
|
|
|
def _c208(db: int):
|
|
"""208(与 SIGNAL_REDIS 同实例)的只读客户端, 按库缓存。"""
|
|
key = ("208", db)
|
|
if key in _clients:
|
|
return _clients[key]
|
|
import redis
|
|
kw = dict(host=settings.SIGNAL_REDIS_HOST, port=settings.SIGNAL_REDIS_PORT,
|
|
password=settings.SIGNAL_REDIS_PASSWORD or None, db=db, decode_responses=True,
|
|
socket_timeout=_to(), socket_connect_timeout=_to())
|
|
try:
|
|
c = redis.Redis(protocol=2, **kw) # RESP2: 服务端 <6.0 不认 HELLO (与行情库同一个坑)
|
|
except TypeError:
|
|
c = redis.Redis(**kw)
|
|
_clients[key] = c
|
|
return c
|
|
|
|
|
|
def _c214():
|
|
"""214(mtf 实况分)的只读客户端, 走 .env 里的 UP_MR_REDIS_URL。"""
|
|
if "214" in _clients:
|
|
return _clients["214"]
|
|
import redis
|
|
url = (getattr(settings, "UP_MR_REDIS_URL", "") or "").strip()
|
|
if not url:
|
|
raise RuntimeError("UP_MR_REDIS_URL 未配置 (请在 .env 加, 见 config/settings.py 注释)")
|
|
try:
|
|
c = redis.Redis.from_url(url, protocol=2, decode_responses=True,
|
|
socket_timeout=_to(), socket_connect_timeout=_to())
|
|
except TypeError:
|
|
c = redis.Redis.from_url(url, decode_responses=True,
|
|
socket_timeout=_to(), socket_connect_timeout=_to())
|
|
_clients["214"] = c
|
|
return c
|
|
|
|
|
|
def _hm(ms):
|
|
"""epoch 毫秒 → HH:MM (本地时区)。拿不到就空串。"""
|
|
try:
|
|
return time.strftime("%H:%M", time.localtime(int(ms) // 1000))
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def _ymd_of(ms):
|
|
"""epoch 毫秒 → YYYY-MM-DD (本地时区)。拿不到就空串 (调用方按「今天」从宽处理)。"""
|
|
try:
|
|
return time.strftime("%Y-%m-%d", time.localtime(int(ms) // 1000))
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def _num(v):
|
|
try:
|
|
return round(float(v), 4)
|
|
except (TypeError, ValueError):
|
|
return v
|
|
|
|
|
|
# ================================================================ 各源解析 (只读)
|
|
def _read_intraday(c, ymd):
|
|
"""一条 intraday_signals 流两家都写: 有 producer_id=intraday_timing / entry_score 的归「择时层 BUY」,
|
|
其余归「决策系统买卖」(bionic 广播 ENTRY/EXIT · BUY/SELL)。"""
|
|
key = "intraday_signals:%s" % ymd
|
|
decision, timing = [], []
|
|
for _id, f in c.xrevrange(key, count=_limit()):
|
|
producer = str(f.get("producer_id") or "")
|
|
is_itd = ("intraday_timing" in producer) or (f.get("entry_score") is not None)
|
|
row = {"ts_code": f.get("ts_code"), "action": f.get("action"),
|
|
"signal_type": f.get("signal_type"), "price": _num(f.get("suggested_price")),
|
|
"target": _num(f.get("target_price")), "confidence": _num(f.get("confidence")),
|
|
"entry_score": _num(f.get("entry_score")), "time": _hm(f.get("trigger_time"))}
|
|
(timing if is_itd else decision).append(row)
|
|
return {"key": key, "decision": decision, "timing": timing}
|
|
|
|
|
|
# 告警源 → (方向, 类型key, 类型标签)。权威依据: intraday_timing/alerts/orchestrator.py 的
|
|
# DIRECTION_DOWN_SOURCES / DIRECTION_UP_SOURCES (2026-06-15 现行) 与《盘中择时层_信号输出说明》§5。
|
|
# · intraday_buy_emitted 是「每发一条 BUY 镜像一条」的事件, **无涨跌方向** (原来按子串误判成看涨)。
|
|
# · multi_source 的方向看 metadata.direction。
|
|
# · volume_capitulation/breakout 已退役(被 capital_distribution/accumulation 取代), 历史键仍可能在流里, 一并登记。
|
|
_ALERT_META = {
|
|
"daily_qrs_symmetric_down": ("看跌", "qrs", "日QRS对称"),
|
|
"money_flow_out_intensity": ("看跌", "money_flow", "资金流强度"),
|
|
"capital_distribution": ("看跌", "capital", "资金分布"),
|
|
"volume_capitulation": ("看跌", "volume", "放量异动"),
|
|
"daily_qrs_symmetric_up": ("看涨", "qrs", "日QRS对称"),
|
|
"money_flow_in_intensity": ("看涨", "money_flow", "资金流强度"),
|
|
"capital_accumulation": ("看涨", "capital", "资金分布"),
|
|
"volume_breakout": ("看涨", "volume", "放量异动"),
|
|
# 股价异动通知: 是"价格大幅变动"事件, 不是情绪信号 → 方向用 涨/跌(区别于看涨/看跌), 单独一类。
|
|
"price_notice_up": ("涨", "price_notice", "股价异动"),
|
|
"price_notice_down": ("跌", "price_notice", "股价异动"),
|
|
"multi_source": (None, "multi", "多源升级"),
|
|
"intraday_buy_emitted": ("—", "buy_emitted", "买入镜像"),
|
|
}
|
|
|
|
|
|
def _read_alerts(c, ymd):
|
|
key = "intraday_alerts:%s" % ymd
|
|
per_cat = _alert_per_cat()
|
|
max_age = _alert_max_age_sec() # 秒; 0=不限龄。超龄不删, 只标 stale 供页面置灰。
|
|
now = time.time()
|
|
counts = {} # cat -> 已收数量; 每类只留最新 per_cat 条(仍按 newest-first)
|
|
out = []
|
|
for _id, f in c.xrevrange(key, count=_alert_scan()):
|
|
src = str(f.get("source") or "")
|
|
meta = f.get("metadata")
|
|
if not isinstance(meta, dict):
|
|
try:
|
|
meta = json.loads(meta)
|
|
except Exception:
|
|
meta = {}
|
|
dir_fixed, cat, cat_label = _ALERT_META.get(src, (None, "other", "其他"))
|
|
if counts.get(cat, 0) >= per_cat: # 该类已满 → 跳过, 高频源不挤掉别的类
|
|
continue
|
|
if dir_fixed is None:
|
|
# multi_source 或未登记源: 方向以 metadata.direction 为准, 拿不到就 —
|
|
mdir = str((meta or {}).get("direction") or "").lower()
|
|
d = "看跌" if mdir in ("down", "short", "bear") else ("看涨" if mdir in ("up", "long", "bull") else "—")
|
|
else:
|
|
d = dir_fixed
|
|
# 分钟龄 + 陈旧标记: 时间窗只做"置灰标注", 不删条目(超龄由页面按 stale 置灰)。
|
|
# trigger_time 是 epoch 毫秒; 缺失/非法 → age_min=None 且不判陈旧(风控从宽, 不误灰)。
|
|
try:
|
|
age_sec = now - float(f.get("trigger_time")) / 1000.0
|
|
age_min = int(age_sec // 60)
|
|
except (TypeError, ValueError):
|
|
age_sec, age_min = None, None
|
|
stale = bool(max_age) and age_sec is not None and age_sec > max_age
|
|
counts[cat] = counts.get(cat, 0) + 1
|
|
out.append({"ts_code": f.get("ts_code"), "source": src, "direction": d,
|
|
"cat": cat, "cat_label": cat_label,
|
|
"level": f.get("level"), "value": _num(f.get("value")),
|
|
"time": _hm(f.get("trigger_time")), "age_min": age_min, "stale": stale})
|
|
return {"key": key, "items": out}
|
|
|
|
|
|
def _read_metrics(c):
|
|
"""mtf 资金异动: 真实载荷打包在 stream 的 data(JSON字符串)里, 顶层只有 ts_code ——
|
|
之前直接取顶层 z_dd/direction 全落空(页面显示为空)。这里拆 data 取真实字段, 顶层作兜底。
|
|
mtf 侧字段: direction(inflow/outflow), z_dd(横截面异动z), window_net(净额), window_ret(区间涨跌); 无原生 level。"""
|
|
key = "mtf:intraday:stream:metrics"
|
|
out = []
|
|
for _id, f in c.xrevrange(key, count=_limit()):
|
|
code = f.get("ts_code") or f.get("code")
|
|
payload = f.get("data")
|
|
if isinstance(payload, str):
|
|
try:
|
|
payload = json.loads(payload)
|
|
except Exception:
|
|
payload = {}
|
|
if not isinstance(payload, dict):
|
|
payload = {}
|
|
g = lambda k, d=None: payload.get(k, f.get(k, d))
|
|
out.append({"ts_code": code, "direction": g("direction"),
|
|
"z_dd": _num(g("z_dd")), "window_net": _num(g("window_net")),
|
|
"window_ret": _num(g("window_ret")), "level": g("level"),
|
|
"value": _num(g("value") if g("value") is not None else g("net")),
|
|
"raw": payload})
|
|
return {"key": key, "items": out}
|
|
|
|
|
|
def _read_sell_actions(c):
|
|
"""风控卖出流: 真实载荷包在 data(JSON 字符串)里, 顶层只有 data 一个字段 ——
|
|
之前直接取顶层 ts_code/confidence/reason 全落空, 页面每行只剩硬编码的 SELL(2026-08-18 修)。
|
|
与消化端 signal_rules.parse_risk_sell 同口径拆 data; 字段名对齐 bionic 写入端
|
|
(tasks_risk/tasks_intraday): 理由是 llm_reason, 时间是 timestamp(epoch 毫秒),
|
|
dominant_signal 区分风控止损与止盈(take_profit)。顶层字段留作兜底, 兼容未来可能的扁平化。
|
|
|
|
**每行带日期与 today 标记** (2026-08-28 页面口径修): 这条流不按日期分键, 清晨没有
|
|
新信号时 xrevrange 取到的「最新 N 条」全是昨天以前的, 而页面只显示 HH:MM —— 昨天
|
|
14:49 的建议卖出会挂在今天 13:20 的页面上冒充新消息。页面据 today 默认只显示今天,
|
|
更早的折叠并标日期。时间戳解析不了的按今天算 (风控从宽, 宁可多显示也不误藏)。"""
|
|
key = "bionic:signals:llm_sell_actions"
|
|
today = time.strftime("%Y-%m-%d")
|
|
out = []
|
|
for _id, f in c.xrevrange(key, count=_limit()):
|
|
raw = f.get("data")
|
|
d = f
|
|
if raw:
|
|
try:
|
|
d = json.loads(raw) if isinstance(raw, str) else dict(raw)
|
|
except Exception:
|
|
d = {}
|
|
ts = d.get("timestamp") or d.get("trigger_time") or f.get("ts")
|
|
ymd = _ymd_of(ts)
|
|
out.append({"ts_code": d.get("ts_code") or f.get("ts_code"),
|
|
"action": d.get("action") or f.get("action"),
|
|
"confidence": _num(d.get("confidence")),
|
|
"dominant_signal": d.get("dominant_signal"),
|
|
"reason": d.get("llm_reason") or d.get("reason") or f.get("reason"),
|
|
"time": _hm(ts), "ymd": ymd,
|
|
"today": (ymd == today) if ymd else True})
|
|
return {"key": key, "items": out}
|
|
|
|
|
|
def _read_mr(c):
|
|
"""mtf 实况分 关注/回避榜 (zset, score=实况分 0~1): 高分=关注、低分=回避, 各取前 10。"""
|
|
key = "mtf:mr:intraday:board"
|
|
top = c.zrevrange(key, 0, 9, withscores=True)
|
|
bottom = c.zrange(key, 0, 9, withscores=True)
|
|
fmt = lambda pairs: [{"ts_code": k, "score": round(float(v), 3)} for k, v in pairs]
|
|
return {"key": key, "watch": fmt(top), "avoid": fmt(bottom)}
|
|
|
|
|
|
# ================================================================ 快照
|
|
def snapshot(force: bool = False) -> dict:
|
|
ttl = max(2, param_store.get_int("PMS_UPSTREAM_CACHE_SEC", 8))
|
|
now = time.time()
|
|
if not force and _cache["data"] is not None and (now - _cache["at"]) < ttl:
|
|
return _cache["data"]
|
|
ymd = datetime.now().strftime("%Y-%m-%d")
|
|
out = {"ok": True, "at": datetime.now().isoformat(timespec="seconds"), "sources": {}}
|
|
|
|
def _try(name, fn):
|
|
try:
|
|
out["sources"][name] = {"ok": True, **fn()}
|
|
except Exception as e:
|
|
logger.warning("[上游信号] %s 读取失败: %s", name, e)
|
|
out["sources"][name] = {"ok": False, "error": f"{type(e).__name__}: {e}"}
|
|
|
|
dbi, dba = settings.SIGNAL_REDIS_DB_INTRADAY, settings.SIGNAL_REDIS_DB_ACTIONS
|
|
_try("intraday", lambda: _read_intraday(_c208(dbi), ymd))
|
|
_try("alerts", lambda: _read_alerts(_c208(dbi), ymd))
|
|
_try("metrics", lambda: _read_metrics(_c208(dbi)))
|
|
_try("sell_actions", lambda: _read_sell_actions(_c208(dba)))
|
|
_try("mr", lambda: _read_mr(_c214()))
|
|
|
|
_cache["at"], _cache["data"] = now, out
|
|
return out
|
|
|
|
# ================================================================ 按代码回扫 (2026-09-09)
|
|
# 信号栏一行 = 上游流里的一条原始消息, 同一只票一天出现很多次 (实测活跃票一天十来条)。
|
|
# 这里按代码把五个源汇到一起, 给页面的「这只票今天的信号」抽屉用。
|
|
#
|
|
# 与 snapshot() 的分工: 那个是**全场最近 N 条**的快照 (告警每类只留最新 20 条), 到下午
|
|
# 一只活跃票早盘的信号早被挤掉了; 这个是**单票回扫到当天开盘**, 所以才叫"全部"。
|
|
# 两份缓存不合并 —— 口径不同, 合并会让单票结果被全场快照污染。
|
|
#
|
|
# 只读纪律与 snapshot() 逐字相同: 一律 XREVRANGE / ZREVRANGE, 不建消费组、不 ACK、不写。
|
|
# **这个接口只能点击触发, 绝不能加进页面轮询** —— 加进去就是每 30 秒对上游做一次全流扫描,
|
|
# 把只读观察变成压力源。
|
|
_BYCODE_CACHE = {}
|
|
_BYCODE_MAX = 32
|
|
_SCAN_PAGE = 500
|
|
|
|
|
|
def _bycode_scan() -> int:
|
|
return max(100, param_store.get_int("PMS_UPSTREAM_BYCODE_SCAN", 5000))
|
|
|
|
|
|
def _code6(v) -> str:
|
|
"""任意写法里抽六位数字。抽不到回空串。"""
|
|
import re
|
|
m = re.search(r"(\d{6})", str(v or ""))
|
|
return m.group(1) if m else ""
|
|
|
|
|
|
def _norm_code(v) -> str:
|
|
"""SH600000 / 600000.SH / sh600000 / 600000 -> 600000.SH; 认不出后缀就只回六位。"""
|
|
s = str(v or "").strip().upper()
|
|
six = _code6(s)
|
|
if not six:
|
|
return ""
|
|
for ex in ("SH", "SZ", "BJ"):
|
|
if ex in s:
|
|
return "%s.%s" % (six, ex)
|
|
return six
|
|
|
|
|
|
def _same_code(a, b) -> bool:
|
|
"""两边都带交易所后缀时要求后缀一致; 只有一边带就退回六位匹配。
|
|
六位在 A 股跨市场唯一, 够用; 这道二次校验是给将来可能出现的别的品种留的。"""
|
|
sa, sb = str(a or "").upper(), str(b or "").upper()
|
|
if _code6(sa) != _code6(sb) or not _code6(sa):
|
|
return False
|
|
exa = next((x for x in ("SH", "SZ", "BJ") if x in sa), "")
|
|
exb = next((x for x in ("SH", "SZ", "BJ") if x in sb), "")
|
|
return (not exa) or (not exb) or exa == exb
|
|
|
|
|
|
def _id_ms(sid) -> int:
|
|
try:
|
|
return int(str(sid).split("-")[0])
|
|
except Exception:
|
|
return 0
|
|
|
|
|
|
def _prev_id(sid) -> str:
|
|
"""比 sid 严格小的边界 id。不用 Redis 6.2 的 `(id` 排他区间 —— 本模块已经因为上游
|
|
可能低于 6.0 而强制走 RESP2, 排他区间同样不能假设。手工减一在所有版本上都成立。"""
|
|
try:
|
|
ms, seq = str(sid).split("-")
|
|
ms, seq = int(ms), int(seq)
|
|
except Exception:
|
|
return "-"
|
|
if seq > 0:
|
|
return "%d-%d" % (ms, seq - 1)
|
|
if ms > 0:
|
|
return "%d-18446744073709551615" % (ms - 1)
|
|
return "-"
|
|
|
|
|
|
def _scan_stream(c, key, *, cap, page=_SCAN_PAGE, stop_before_ms=None):
|
|
"""分页倒序回扫一条流, 返回 (条目列表, 是否还没见底)。
|
|
|
|
一次 XREVRANGE COUNT 5000 会把几兆塞进一个应答且没法中途停, 所以分页。
|
|
stop_before_ms 给**不按日期分键**的流用 (风控卖出、资金异动): 一旦某条早于今天零点
|
|
就立刻收手, 且不算截断 —— 今天的已经扫全了。
|
|
"""
|
|
entries, cur, truncated = [], "+", False
|
|
while len(entries) < cap:
|
|
want = min(page, cap - len(entries))
|
|
batch = c.xrevrange(key, max=cur, min="-", count=want)
|
|
if not batch:
|
|
break
|
|
stop = False
|
|
for sid, f in batch:
|
|
if stop_before_ms is not None and _id_ms(sid) < stop_before_ms:
|
|
stop = True
|
|
break
|
|
entries.append((sid, f))
|
|
if stop:
|
|
return entries, False
|
|
if len(batch) < want:
|
|
break
|
|
cur = _prev_id(batch[-1][0])
|
|
if cur == "-":
|
|
break
|
|
else:
|
|
truncated = True
|
|
return entries, truncated
|
|
|
|
|
|
def _today_start_ms() -> int:
|
|
t = time.localtime()
|
|
return int(time.mktime((t.tm_year, t.tm_mon, t.tm_mday, 0, 0, 0, 0, 0, -1)) * 1000)
|
|
|
|
|
|
def _row(src, src_label, ms, **kw) -> dict:
|
|
"""时间线一行的统一形状。缺的字段给 None 不缺键 —— 模板最怕键时有时无。"""
|
|
base = {"src": src, "src_label": src_label, "ts": int(ms or 0), "time": _hm(ms),
|
|
"ymd": _ymd_of(ms), "cat": None, "cat_label": None, "direction": None,
|
|
"level": None, "value": None, "price": None, "target": None, "confidence": None,
|
|
"entry_score": None, "z_dd": None, "window_net": None, "window_ret": None,
|
|
"action": None, "dominant_signal": None, "reason": None, "stale": False, "flow": None,
|
|
"age_min": None, "dup": 1, "ts_from": "trigger"}
|
|
base.update(kw)
|
|
return base
|
|
|
|
|
|
def _dedup(rows) -> list:
|
|
"""同源 + 同分钟 + 同值的合并成一条并累加 dup; **跨源永不合并** ——
|
|
决策系统与择时层在同一分钟都看多, 是两条独立证据, 合并就是删信息。"""
|
|
seen, out = {}, []
|
|
for r in rows:
|
|
ident = {
|
|
"alert": (r.get("cat"), r.get("level"), r.get("value")),
|
|
"decision": (r.get("action"), r.get("price")),
|
|
"timing": (r.get("price"), r.get("entry_score")),
|
|
"sell": (r.get("dominant_signal"), r.get("confidence")),
|
|
"metrics": (r.get("z_dd"), r.get("window_net")),
|
|
}.get(r.get("src"), (r.get("value"),))
|
|
key = (r.get("src"), r.get("time"), ident)
|
|
if key in seen:
|
|
seen[key]["dup"] += 1
|
|
continue
|
|
seen[key] = r
|
|
out.append(r)
|
|
return out
|
|
|
|
|
|
def _bycode_intraday(c, ymd, code, cap):
|
|
ent, trunc = _scan_stream(c, "intraday_signals:%s" % ymd, cap=cap)
|
|
rows = []
|
|
for _sid, f in ent:
|
|
if not _same_code(f.get("ts_code"), code):
|
|
continue
|
|
producer = str(f.get("producer_id") or "")
|
|
is_itd = ("intraday_timing" in producer) or (f.get("entry_score") is not None)
|
|
rows.append(_row("timing" if is_itd else "decision",
|
|
"盘中择时层入场" if is_itd else "决策系统广播",
|
|
f.get("trigger_time"),
|
|
cat=f.get("signal_type"),
|
|
cat_label="择时层入场" if is_itd else ("建议买入" if str(f.get("action") or "").upper() == "BUY" else "建议卖出"),
|
|
direction="看涨" if str(f.get("action") or "").upper() == "BUY" else "看跌",
|
|
action=f.get("action"), price=_num(f.get("suggested_price")),
|
|
target=_num(f.get("target_price")), confidence=_num(f.get("confidence")),
|
|
entry_score=_num(f.get("entry_score"))))
|
|
return rows, len(ent), trunc
|
|
|
|
|
|
def _bycode_alerts(c, ymd, code, cap):
|
|
ent, trunc = _scan_stream(c, "intraday_alerts:%s" % ymd, cap=cap)
|
|
max_age, now, rows = _alert_max_age_sec(), time.time(), []
|
|
for _sid, f in ent:
|
|
if not _same_code(f.get("ts_code"), code):
|
|
continue
|
|
src = str(f.get("source") or "")
|
|
meta = f.get("metadata")
|
|
if not isinstance(meta, dict):
|
|
try:
|
|
meta = json.loads(meta)
|
|
except Exception:
|
|
meta = {}
|
|
dir_fixed, cat, cat_label = _ALERT_META.get(src, (None, "other", "其他"))
|
|
if dir_fixed is None:
|
|
mdir = str((meta or {}).get("direction") or "").lower()
|
|
d = "看跌" if mdir in ("down", "short", "bear") else ("看涨" if mdir in ("up", "long", "bull") else "—")
|
|
else:
|
|
d = dir_fixed
|
|
try:
|
|
age_sec = now - float(f.get("trigger_time")) / 1000.0
|
|
age_min = int(age_sec // 60)
|
|
except (TypeError, ValueError):
|
|
age_sec, age_min = None, None
|
|
rows.append(_row("alert", "盘中告警", f.get("trigger_time"), cat=cat, cat_label=cat_label,
|
|
direction=d, level=f.get("level"), value=_num(f.get("value")),
|
|
age_min=age_min,
|
|
stale=bool(max_age) and age_sec is not None and age_sec > max_age,
|
|
reason=str(f.get("source") or "")))
|
|
return rows, len(ent), trunc
|
|
|
|
|
|
def _bycode_sell(c, code, cap):
|
|
ent, trunc = _scan_stream(c, "bionic:signals:llm_sell_actions", cap=cap,
|
|
stop_before_ms=_today_start_ms())
|
|
rows = []
|
|
for _sid, f in ent:
|
|
raw = f.get("data")
|
|
d = f
|
|
if raw:
|
|
try:
|
|
d = json.loads(raw) if isinstance(raw, str) else dict(raw)
|
|
except Exception:
|
|
d = {}
|
|
if not _same_code(d.get("ts_code") or f.get("ts_code"), code):
|
|
continue
|
|
ts = d.get("timestamp") or d.get("trigger_time") or f.get("ts")
|
|
rows.append(_row("sell", "风控卖出动作", ts, cat="risk_sell", cat_label="风控卖出",
|
|
direction="看跌", action=d.get("action"),
|
|
confidence=_num(d.get("confidence")),
|
|
dominant_signal=d.get("dominant_signal"),
|
|
reason=d.get("llm_reason") or d.get("reason") or f.get("reason")))
|
|
return rows, len(ent), trunc
|
|
|
|
|
|
def _bycode_metrics(c, code, cap):
|
|
"""资金异动进时间线 (2026-09-09): 时间取流 id 的毫秒部分, 那是**写入时刻**不是上游触发时刻,
|
|
所以每行标 ts_from=stream_id, 页面写「写入 13:41」而不是「触发 13:41」。"""
|
|
ent, trunc = _scan_stream(c, "mtf:intraday:stream:metrics", cap=cap,
|
|
stop_before_ms=_today_start_ms())
|
|
rows = []
|
|
for sid, f in ent:
|
|
cd = f.get("ts_code") or f.get("code")
|
|
payload = f.get("data")
|
|
if isinstance(payload, str):
|
|
try:
|
|
payload = json.loads(payload)
|
|
except Exception:
|
|
payload = {}
|
|
if not isinstance(payload, dict):
|
|
payload = {}
|
|
if not _same_code(cd or payload.get("ts_code"), code):
|
|
continue
|
|
g = lambda k, dv=None: payload.get(k, f.get(k, dv))
|
|
_d = str(g("direction") or "").lower()
|
|
rows.append(_row("metrics", "盘中资金异动", _id_ms(sid), cat="fund_flow", cat_label="资金异动",
|
|
direction={"inflow": "看涨", "outflow": "看跌"}.get(_d, g("direction")),
|
|
flow=_d or None, level=g("level"),
|
|
value=_num(g("value") if g("value") is not None else g("net")),
|
|
z_dd=_num(g("z_dd")), window_net=_num(g("window_net")),
|
|
window_ret=_num(g("window_ret")), ts_from="stream_id"))
|
|
return rows, len(ent), trunc
|
|
|
|
|
|
def by_code(ts_code: str, force: bool = False) -> dict:
|
|
"""一只票今天的全部上游信号。点击触发, 不进任何轮询。"""
|
|
code = _norm_code(ts_code)
|
|
if not code:
|
|
return {"ok": False, "error": "认不出的股票代码: %s" % ts_code, "timeline": [],
|
|
"snapshot": {"metrics": [], "mr": {"watch": None, "avoid": None}}}
|
|
six = _code6(code)
|
|
ttl = max(2, param_store.get_int("PMS_UPSTREAM_CACHE_SEC", 8))
|
|
now = time.time()
|
|
hit = _BYCODE_CACHE.get(six)
|
|
if not force and hit and (now - hit[0]) < ttl:
|
|
return hit[1]
|
|
|
|
ymd = datetime.now().strftime("%Y-%m-%d")
|
|
cap = _bycode_scan()
|
|
out = {"ok": True, "ts_code": code, "code6": six, "ymd": ymd,
|
|
"at": datetime.now().isoformat(timespec="seconds"),
|
|
"timeline": [], "snapshot": {"metrics": [], "mr": {"watch": None, "avoid": None}},
|
|
"scanned": {}, "truncated": {}, "sources": {}}
|
|
rows = []
|
|
|
|
def _try(name, fn):
|
|
try:
|
|
got, scanned, trunc = fn()
|
|
rows.extend(got)
|
|
out["scanned"][name] = scanned
|
|
out["truncated"][name] = bool(trunc)
|
|
out["sources"][name] = {"ok": True}
|
|
except Exception as e:
|
|
logger.warning("[单票信号] %s %s 读取失败: %s", code, name, e)
|
|
out["sources"][name] = {"ok": False, "error": "%s: %s" % (type(e).__name__, e)}
|
|
|
|
dbi, dba = settings.SIGNAL_REDIS_DB_INTRADAY, settings.SIGNAL_REDIS_DB_ACTIONS
|
|
_try("intraday", lambda: _bycode_intraday(_c208(dbi), ymd, code, cap))
|
|
_try("alerts", lambda: _bycode_alerts(_c208(dbi), ymd, code, cap))
|
|
_try("sell_actions", lambda: _bycode_sell(_c208(dba), code, cap))
|
|
_try("metrics", lambda: _bycode_metrics(_c208(dbi), code, cap))
|
|
|
|
# 实况分是一个按分数排的有序集合, 连写入时刻都没有 —— **绝不给它编时间**,
|
|
# 排进时间线的假时间戳比放在快照段里说"没有时间"危险得多。
|
|
try:
|
|
mr = _read_mr(_c214())
|
|
out["snapshot"]["mr"] = {
|
|
"watch": next((x for x in mr.get("watch") or [] if _same_code(x.get("ts_code"), code)), None),
|
|
"avoid": next((x for x in mr.get("avoid") or [] if _same_code(x.get("ts_code"), code)), None)}
|
|
out["sources"]["mr"] = {"ok": True}
|
|
except Exception as e:
|
|
logger.warning("[单票信号] %s mr 读取失败: %s", code, e)
|
|
out["sources"]["mr"] = {"ok": False, "error": "%s: %s" % (type(e).__name__, e)}
|
|
|
|
# 资金异动同时进时间线与快照段: 时间线给"什么时候进的钱", 快照段给"现在是什么状态"。
|
|
out["snapshot"]["metrics"] = [r for r in rows if r["src"] == "metrics"][:5]
|
|
rows = _dedup(rows)
|
|
rows.sort(key=lambda r: (r.get("ts") or 0), reverse=True)
|
|
out["timeline"] = rows
|
|
|
|
_BYCODE_CACHE[six] = (now, out)
|
|
if len(_BYCODE_CACHE) > _BYCODE_MAX:
|
|
for k in sorted(_BYCODE_CACHE, key=lambda x: _BYCODE_CACHE[x][0])[:-_BYCODE_MAX]:
|
|
_BYCODE_CACHE.pop(k, None)
|
|
return out
|