mirror of
https://ghfast.top/https://github.com/aeroxw/tick-stock-panel.git
synced 2026-09-12 21:24:16 +08:00
shy3130
d89b9068a7
连板梯队: - 新增涨停/跌停方向切换(胶囊式, 涨停红/跌停绿) - 跌停侧三状态: 跌停/翘板/止跌(对称涨停侧) - 涨跌停双计数 + 真假板修正/降级标识(问号弹窗) 五档盘口 sealed(真假涨停): - depth_service 独立旁路线(只读enriched, 不动计算逻辑) - 涨跌停一体(一次depth.batch双向覆盖) - 假涨停归炸板/假跌停归翘板, 真封板显示封单量 - 三层防护节流(套餐clamp+限速clamp+接管通知) - Pro/Expert盘中轮询 + 盘后定版job + 启动补跑 - 收盘job可配置(15:01~18:00, 默认15:02) - 设置页'连板梯队降级修正'卡片(开关+配置+立即修正) - 看板页同步应用sealed修正+降级标识 配置/SDK: - tiers.yaml拆分depth5/depth5.batch, 补全quote.batch - tickflow SDK升级0.1.21→0.1.23(depth.batch可用) - 实时行情轮询间隔下限调整(pro 2s/expert 1s) - capabilities缓存schema版本化(v3) 其他: - 档位配色抽共享组件TierTag/tierStyle - Keys页档位问号弹窗(4档位tag+检测说明) - git hooks自动版本号+0.0.1+打tag
77 lines
2.5 KiB
Python
77 lines
2.5 KiB
Python
"""Capability 定义(§5.1)。
|
|
|
|
业务代码只依赖 CapabilitySet,不读 tiers.yaml,不感知"档位"。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from enum import StrEnum
|
|
|
|
|
|
class Cap(StrEnum):
|
|
"""所有 capability 的命名常量。新增能力时只在这里加一行。"""
|
|
|
|
QUOTE_BY_SYMBOL = "quote.by_symbol"
|
|
QUOTE_BATCH = "quote.batch"
|
|
QUOTE_POOL = "quote.pool"
|
|
KLINE_DAILY_BY_SYMBOL = "kline.daily.by_symbol"
|
|
KLINE_DAILY_BATCH = "kline.daily.batch"
|
|
KLINE_MINUTE_BY_SYMBOL = "kline.minute.by_symbol"
|
|
KLINE_MINUTE_BATCH = "kline.minute.batch"
|
|
INTRADAY = "intraday"
|
|
INTRADAY_BATCH = "intraday.batch"
|
|
DEPTH5 = "depth5"
|
|
DEPTH5_BATCH = "depth5.batch"
|
|
WEBSOCKET = "websocket"
|
|
FINANCIAL = "financial"
|
|
ADJ_FACTOR = "adj_factor"
|
|
|
|
|
|
@dataclass(slots=True, frozen=True)
|
|
class CapabilityLimits:
|
|
"""单个 capability 的运行时限制。"""
|
|
rpm: int | None = None # 次/分钟,None 表示未知或不限
|
|
batch: int | None = None # 标的/次
|
|
subscribe: int | None = None # WS 订阅上限
|
|
|
|
|
|
class CapabilitySet:
|
|
"""探测得到的"用户当前可用能力"。业务代码的唯一真理源。"""
|
|
|
|
def __init__(self, caps: dict[Cap, CapabilityLimits] | None = None) -> None:
|
|
self._caps: dict[Cap, CapabilityLimits] = dict(caps or {})
|
|
|
|
def has(self, cap: Cap) -> bool:
|
|
return cap in self._caps
|
|
|
|
def limits(self, cap: Cap) -> CapabilityLimits | None:
|
|
return self._caps.get(cap)
|
|
|
|
def require(self, cap: Cap) -> CapabilityLimits:
|
|
"""断言可用,否则抛 CapabilityDenied。"""
|
|
if cap not in self._caps:
|
|
raise CapabilityDenied(cap)
|
|
return self._caps[cap]
|
|
|
|
def all(self) -> dict[Cap, CapabilityLimits]:
|
|
return dict(self._caps)
|
|
|
|
def to_dict(self) -> dict[str, dict]:
|
|
return {
|
|
str(cap): {
|
|
"rpm": lim.rpm,
|
|
"batch": lim.batch,
|
|
"subscribe": lim.subscribe,
|
|
}
|
|
for cap, lim in self._caps.items()
|
|
}
|
|
|
|
|
|
class CapabilityDenied(Exception):
|
|
"""请求的 capability 当前不可用。"""
|
|
|
|
def __init__(self, cap: Cap, suggestion: str | None = None) -> None:
|
|
self.cap = cap
|
|
self.suggestion = suggestion or f"加购『{cap}』能力可解锁"
|
|
super().__init__(f"capability not available: {cap}; {self.suggestion}")
|