Merge remote-tracking branch 'origin/main' into v0.2

This commit is contained in:
shy3130
2026-08-10 12:37:06 +08:00
14 changed files with 1214 additions and 30 deletions
+46 -1
View File
@@ -59,15 +59,34 @@ class ConditionModel(BaseModel):
value: float | None = None # op 非 truth 时必填
class SectorTargetModel(BaseModel):
key: str
kind: str
name: str
symbol: str | None = None
source_id: str | None = None
field: str | None = None
source_field: str | None = None
value: str | None = None
level: int | None = None
available: bool = True
member_count: int = 0
class RuleModel(BaseModel):
id: str
name: str
enabled: bool = True
type: str # strategy | signal | price | market
type: str # strategy | signal | price | market | sector
asset_type: str = "stock" # stock | etf (etf: strategy 型走 ETF 历史加载器)
scope: str = "symbols" # symbols | all | sector
symbols: list[str] = []
sector: str | None = None
sector_kind: str | None = None # index | concept | industry
sector_targets: list[SectorTargetModel] = []
sector_trigger: str = "change_pct" # change_pct | momentum
threshold_pct: float = 1.0
window_minutes: int = 5
strategy_id: str | None = None
direction: str = "entry" # entry | exit | both
notify_events: list[str] | None = None
@@ -119,6 +138,10 @@ def get_options(request: Request):
except Exception:
pass
sector_service = getattr(request.app.state, "sector_monitor_service", None)
sector_targets = sector_service.list_targets() if sector_service is not None else {
"index": [], "concept": [], "industry": [],
}
return {
"threshold_fields": threshold_fields,
"builtin_signals": builtin_signals,
@@ -129,6 +152,7 @@ def get_options(request: Request):
{"key": "price", "label": "价格/涨跌"},
{"key": "market", "label": "市场异动"},
{"key": "strategy", "label": "策略监控"},
{"key": "sector", "label": "板块监控"},
],
"scopes": [
{"key": "symbols", "label": "指定标的"},
@@ -152,6 +176,7 @@ def get_options(request: Request):
"intraday_signal_support": intraday_monitor_support(
getattr(request.app.state, "capabilities", None),
),
"sector_targets": sector_targets,
}
@@ -186,6 +211,17 @@ def list_rules(request: Request):
if runtime_warning:
for rule in intraday_rules:
rule["runtime_warning"] = runtime_warning
sector_service = getattr(request.app.state, "sector_monitor_service", None)
if sector_service is not None:
for rule in rules:
if rule.get("type") != "sector":
continue
missing = sector_service.missing_target_keys(rule.get("sector_targets", []))
unavailable = sector_service.unavailable_target_keys(rule.get("sector_targets", []))
if missing:
rule["runtime_warning"] = "部分板块数据已不存在, 请重新选择监控对象"
elif unavailable:
rule["runtime_warning"] = "所选指数未加入实时指数池, 请先在实时监控设置中启用"
# 按 created_at 倒序
rules.sort(key=lambda r: r.get("created_at", ""), reverse=True)
return {"rules": rules}
@@ -232,6 +268,15 @@ def save_rule(req: RuleModel, request: Request):
monitor_rules.validate(rule)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
if rule.get("type") == "sector":
sector_service = getattr(request.app.state, "sector_monitor_service", None)
if sector_service is None:
raise HTTPException(status_code=503, detail="板块监控服务未初始化")
targets = rule.get("sector_targets", [])
if sector_service.missing_target_keys(targets):
raise HTTPException(status_code=400, detail="所选板块数据已变化, 请重新选择")
if sector_service.unavailable_target_keys(targets):
raise HTTPException(status_code=400, detail="所选指数未加入实时指数池, 请先在实时监控设置中启用")
if rule.get("enabled", True) and uses_intraday_signals(rule):
from app.services.kline_sync import intraday_monitor_support
+4
View File
@@ -214,9 +214,12 @@ async def lifespan(app: FastAPI):
from app.strategy.monitor import MonitorRuleEngine
from app.strategy import monitor_rules as mr_store
from app.services import preferences
from app.services.sector_monitor import SectorMonitorService
monitor_engine = MonitorRuleEngine()
sector_monitor_service = SectorMonitorService(repo)
monitor_engine.set_strategy_engine(strategy_engine)
monitor_engine.set_data_dir(store.data_dir)
monitor_engine.set_sector_monitor_service(sector_monitor_service)
# 复用 ScreenerService 的历史窗口加载器 (三级缓存, 启动预计算命中 ~0ms),
# 让声明 filter_history 的策略 (如反包) 也能在实时监控里跑选股 → 盘中触发通知。
monitor_engine.set_history_loader(_screener_svc._load_enriched_history)
@@ -241,6 +244,7 @@ async def lifespan(app: FastAPI):
except Exception as e: # noqa: BLE001
logger.warning("monitor engine load failed: %s", e)
app.state.monitor_engine = monitor_engine
app.state.sector_monitor_service = sector_monitor_service
yield
+18 -3
View File
@@ -1116,6 +1116,11 @@ class QuoteService:
rule_events = engine.evaluate(eval_df, asset_type="stock")
if engine.consume_strategy_result_updates():
self.notify_strategy_results_updated()
if engine.has_rule_type("sector"):
rule_events += engine.evaluate_sectors(
enriched_today if stock_ready else pl.DataFrame(),
self.get_index_quotes(),
)
# ETF 规则轮: 股票快照不含 ETF, 用 ETF enriched 快照单独评估。
# 独立 try —— ETF 轮任何异常都不得丢弃本轮已算出的股票告警。
# refresh=False —— 不在轮询线程上触发 ETF 冷缓存的同步重算 (缓存由 ETF 实时
@@ -1154,7 +1159,7 @@ class QuoteService:
logger.warning("告警落盘失败: %s", e)
# 转为 SSE 推送格式 (兼容旧 alert schema)
for ev in rule_events:
all_alerts.append({
alert = {
"source": ev["source"],
"type": ev["type"],
"rule_id": ev.get("rule_id"),
@@ -1168,7 +1173,16 @@ class QuoteService:
"severity": ev.get("severity", "info"),
"conditions": ev.get("conditions") or [],
"logic": ev.get("logic") or "and",
})
}
for key in (
"sector_kind", "sector_key", "sector_name",
"sector_source_field", "sector_value", "sector_level",
"window_change_pct", "coverage_ratio", "valid_count",
"total_count", "up_count", "down_count", "leader",
):
if key in ev:
alert[key] = ev[key]
all_alerts.append(alert)
# 策略页实时回显: 不写文件 (实时行情每轮更新 enriched, 写文件会被 read_cache
# 的 mtime 校验判过期, 反复读不到)。监控引擎本轮已算出的结果存在内存
@@ -1339,6 +1353,7 @@ class QuoteService:
source_labels = {
"strategy": "策略", "signal": "信号",
"price": "价格", "market": "异动", "ladder": "连板梯队",
"sector": "板块",
}
rules = engine.rules if engine is not None else {}
enqueued = 0
@@ -1391,7 +1406,7 @@ class QuoteService:
source = ev.get("source", "")
source_label = {
"strategy": "策略", "signal": "信号",
"price": "价格", "market": "异动",
"price": "价格", "market": "异动", "sector": "板块",
}.get(source, source or "通知")
name = ev.get("name") or ""
+361
View File
@@ -0,0 +1,361 @@
"""板块监控目标目录与实时聚合快照。"""
from __future__ import annotations
import hashlib
import math
import re
from collections import deque
from datetime import datetime
from pathlib import Path
from typing import Any
import polars as pl
from app.services import preferences
from app.services.ext_data import ExtConfig, ExtConfigStore
CORE_INDICES = {
"000001.SH": "上证指数",
"399001.SZ": "深证成指",
"399006.SZ": "创业板指",
"000680.SH": "科创综指",
}
SECTOR_KINDS = {"index", "concept", "industry"}
_VALUE_SEP = re.compile(r"[\u3001,\uff0c;\uff1b|]+")
_NULL_VALUES = {"nan", "none", "null", "<na>", "n/a", "-"}
_HISTORY_SECONDS = 31 * 60
_WINDOW_TOLERANCE_SECONDS = 90
def _finite(value: Any) -> float | None:
if value is None:
return None
try:
number = float(value)
except (TypeError, ValueError):
return None
return number if math.isfinite(number) else None
def _dimension_kind(field_name: str, field_label: str) -> str | None:
text = f"{field_name} {field_label}".lower()
if any(word in text for word in ("概念", "题材", "concept", "theme")):
return "concept"
if any(word in text for word in ("行业", "申万", "中信", "industry", "sector")):
return "industry"
return None
def _target_key(kind: str, source_id: str, field: str, value: str, level: int | None) -> str:
raw = f"{kind}\0{source_id}\0{field}\0{level or 0}\0{value}"
digest = hashlib.sha1(raw.encode("utf-8")).hexdigest()[:16]
return f"{kind}:{digest}"
class SectorMonitorService:
"""缓存板块成员关系, 并按启用规则构建轻量实时快照。"""
def __init__(self, repo) -> None:
self._repo = repo
self._data_dir: Path = repo.store.data_dir
self._catalog_signature: tuple[tuple[str, int, int], ...] | None = None
self._catalog: dict[str, list[dict]] = {kind: [] for kind in SECTOR_KINDS}
self._targets_by_key: dict[str, dict] = {}
self._members_by_key: dict[str, set[str]] = {}
self._history: dict[str, deque[tuple[float, float]]] = {}
self._history_day: str | None = None
def list_targets(self) -> dict[str, list[dict]]:
self._ensure_catalog()
return {kind: [dict(item) for item in self._catalog[kind]] for kind in self._catalog}
def missing_target_keys(self, targets: list[dict]) -> list[str]:
self._ensure_catalog()
return [str(target.get("key") or "") for target in targets if target.get("key") not in self._targets_by_key]
def unavailable_target_keys(self, targets: list[dict]) -> list[str]:
self._ensure_catalog()
return [
str(target.get("key") or "")
for target in targets
if target.get("key") in self._targets_by_key
and not self._targets_by_key[target["key"]].get("available", True)
]
def build_snapshots(
self,
stock_df: pl.DataFrame,
index_df: pl.DataFrame,
targets: list[dict],
windows: set[int],
*,
now: float,
) -> dict[str, dict]:
if not targets:
return {}
self._ensure_catalog()
self._reset_history_for_day(now)
stock_rows = self._row_map(stock_df, index_values_are_percent=False)
index_rows = self._row_map(index_df, index_values_are_percent=True)
snapshots: dict[str, dict] = {}
for raw_target in targets:
key = str(raw_target.get("key") or "")
target = self._targets_by_key.get(key)
if not target:
continue
if target["kind"] == "index":
snapshot = self._index_snapshot(target, index_rows)
else:
snapshot = self._dimension_snapshot(target, stock_rows)
if snapshot is None:
continue
change_pct = snapshot.get("change_pct")
history = self._history.setdefault(key, deque())
if snapshot["valid"] and change_pct is not None:
history.append((now, float(change_pct)))
while history and history[0][0] < now - _HISTORY_SECONDS:
history.popleft()
snapshot["window_changes"] = {
window: self._window_change(history, now, window, change_pct)
for window in windows
}
snapshots[key] = snapshot
return snapshots
def _ensure_catalog(self) -> None:
signature = self._data_signature()
if signature == self._catalog_signature:
return
catalog = {kind: [] for kind in SECTOR_KINDS}
targets_by_key: dict[str, dict] = {}
members_by_key: dict[str, set[str]] = {}
index_names = dict(CORE_INDICES)
try:
indices = self._repo.get_index_instruments()
if not indices.is_empty() and "symbol" in indices.columns:
for row in indices.to_dicts():
if row.get("asset_type") == "etf":
continue
symbol = str(row.get("symbol") or "").strip().upper()
if symbol:
index_names[symbol] = str(row.get("name") or symbol)
except Exception:
pass
realtime_index_enabled = preferences.get_realtime_pull_index()
realtime_indices = set(preferences.get_realtime_index_symbols() or CORE_INDICES)
all_indices_enabled = preferences.get_realtime_index_mode() == "all"
for symbol, name in sorted(index_names.items()):
target = {
"key": f"index:{symbol}",
"kind": "index",
"name": name,
"symbol": symbol,
"available": realtime_index_enabled and (all_indices_enabled or symbol in realtime_indices),
"member_count": 1,
}
catalog["index"].append(target)
targets_by_key[target["key"]] = target
catalog["index"].sort(key=lambda item: (not item["available"], item["symbol"]))
for config in ExtConfigStore(self._data_dir).load_all():
df = self._read_ext_dataframe(config)
if df.is_empty():
continue
symbol_col = self._symbol_column(config, df)
if not symbol_col:
continue
for field in config.fields:
kind = _dimension_kind(field.name, field.label)
if kind is None or field.name not in df.columns:
continue
for row in df.select([symbol_col, field.name]).iter_rows(named=True):
symbol = str(row.get(symbol_col) or "").strip().upper()
if not symbol:
continue
for raw_value in self._dimension_values(row.get(field.name)):
paths = self._industry_paths(raw_value) if kind == "industry" else [(raw_value, None, raw_value)]
for value, level, name in paths:
key = _target_key(kind, config.id, field.name, value, level)
members_by_key.setdefault(key, set()).add(symbol)
if key not in targets_by_key:
target = {
"key": key,
"kind": kind,
"name": name,
"source_id": config.id,
"field": field.name,
"source_field": f"{config.id}.{field.name}",
"value": value,
"level": level,
"available": True,
}
targets_by_key[key] = target
catalog[kind].append(target)
for kind in ("concept", "industry"):
for target in catalog[kind]:
target["member_count"] = len(members_by_key.get(target["key"], set()))
catalog[kind].sort(key=lambda item: (item.get("level") or 0, item["name"], item["value"]))
self._catalog_signature = signature
self._catalog = catalog
self._targets_by_key = targets_by_key
self._members_by_key = members_by_key
self._history.clear()
def _data_signature(self) -> tuple[tuple[str, int, int], ...]:
base = self._data_dir / "ext_data"
paths: list[Path] = []
for config in ExtConfigStore(self._data_dir).load_all():
if not any(_dimension_kind(field.name, field.label) for field in config.fields):
continue
config_dir = base / config.id
paths.extend(config_dir.rglob("config.json"))
paths.extend(config_dir.rglob("*.parquet"))
signature = [
(str(path), path.stat().st_mtime_ns, path.stat().st_size)
for path in sorted(paths)
if path.is_file()
]
index_mode = preferences.get_realtime_index_mode()
index_enabled = preferences.get_realtime_pull_index()
index_symbols = sorted(preferences.get_realtime_index_symbols() or CORE_INDICES)
signature.append((f"realtime_indices:{index_enabled}:{index_mode}:{','.join(index_symbols)}", 0, 0))
return tuple(signature)
def _read_ext_dataframe(self, config: ExtConfig) -> pl.DataFrame:
base = self._data_dir / "ext_data" / config.id
if config.mode == "timeseries":
files = sorted((base / "timeseries").rglob("*.parquet"))
files = files[-1:] if files else []
else:
files = sorted(base.glob("*.parquet"))
if not files:
return pl.DataFrame()
try:
return pl.read_parquet(files)
except Exception:
return pl.DataFrame()
@staticmethod
def _symbol_column(config: ExtConfig, df: pl.DataFrame) -> str | None:
candidates = ["symbol", "code", "股票代码", "代码"]
for mapping in (config.symbol_map, config.code_map):
if isinstance(mapping, dict) and mapping.get("type") == "mapped":
candidates.append(str(mapping.get("col") or ""))
return next((column for column in candidates if column in df.columns), None)
@staticmethod
def _dimension_values(raw: Any) -> list[str]:
if raw is None:
return []
return [
value.strip()
for value in _VALUE_SEP.split(str(raw))
if value.strip() and value.strip().casefold() not in _NULL_VALUES
]
@staticmethod
def _industry_paths(raw: str) -> list[tuple[str, int, str]]:
parts = [part.strip() for part in raw.split("-") if part.strip()]
if not parts:
return []
return [
("-".join(parts[:level]), level, " / ".join(parts[:level]))
for level in range(1, len(parts) + 1)
]
@staticmethod
def _row_map(df: pl.DataFrame, *, index_values_are_percent: bool) -> dict[str, dict]:
if df.is_empty() or "symbol" not in df.columns:
return {}
rows: dict[str, dict] = {}
for row in df.to_dicts():
symbol = str(row.get("symbol") or "").strip().upper()
if not symbol:
continue
change_pct = _finite(row.get("change_pct"))
if change_pct is not None and index_values_are_percent:
change_pct /= 100
rows[symbol] = {**row, "change_pct": change_pct}
return rows
@staticmethod
def _index_snapshot(target: dict, index_rows: dict[str, dict]) -> dict | None:
row = index_rows.get(str(target.get("symbol") or "").upper())
if not row or row.get("change_pct") is None:
return None
return {
**target,
"valid": True,
"change_pct": row["change_pct"],
"price": _finite(row.get("close") or row.get("last_price")),
"coverage_ratio": 1.0,
"valid_count": 1,
"total_count": 1,
"up_count": int(row["change_pct"] > 0),
"down_count": int(row["change_pct"] < 0),
"leader": None,
}
def _dimension_snapshot(self, target: dict, stock_rows: dict[str, dict]) -> dict | None:
members = self._members_by_key.get(target["key"], set())
if not members:
return None
valid_rows = [
stock_rows[symbol]
for symbol in members
if symbol in stock_rows and stock_rows[symbol].get("change_pct") is not None
]
total_count = len(members)
valid_count = len(valid_rows)
coverage_ratio = valid_count / total_count if total_count else 0.0
valid = total_count >= 5 and coverage_ratio >= 0.8
changes = [float(row["change_pct"]) for row in valid_rows]
leader = max(valid_rows, key=lambda row: row["change_pct"]) if valid_rows else None
return {
**target,
"valid": valid,
"change_pct": sum(changes) / len(changes) if changes else None,
"price": None,
"coverage_ratio": coverage_ratio,
"valid_count": valid_count,
"total_count": total_count,
"up_count": sum(value > 0 for value in changes),
"down_count": sum(value < 0 for value in changes),
"leader": {
"symbol": leader.get("symbol"),
"name": leader.get("name"),
"change_pct": leader.get("change_pct"),
} if leader else None,
}
@staticmethod
def _window_change(
history: deque[tuple[float, float]],
now: float,
window: int,
current: float | None,
) -> float | None:
if current is None:
return None
cutoff = now - window * 60
for timestamp, previous in reversed(history):
if timestamp <= cutoff:
if timestamp < cutoff - _WINDOW_TOLERANCE_SECONDS:
return None
return current - previous
return None
def _reset_history_for_day(self, now: float) -> None:
day = datetime.fromtimestamp(now).date().isoformat()
if self._history_day == day:
return
self._history_day = day
self._history.clear()
+172
View File
@@ -353,6 +353,8 @@ class MonitorRuleEngine:
self._building_strategy_results: dict[str, dict] = {}
# 本轮成功写入股票策略实时结果的策略 ID, 供 QuoteService 在计算完成后精确通知策略页。
self._latest_strategy_result_ids: set[str] = set()
self._sector_monitor_service = None
self._sector_condition_state: dict[tuple[str, str], bool] = {}
def set_strategy_engine(self, engine) -> None:
"""注入 StrategyEngine, type=strategy 规则据此跑选股。"""
@@ -362,6 +364,9 @@ class MonitorRuleEngine:
"""注入数据目录, 用于加载策略的用户覆盖配置。"""
self._data_dir = data_dir
def set_sector_monitor_service(self, service) -> None:
self._sector_monitor_service = service
def invalidate_strategy_state(self) -> None:
"""策略注册表变更后清除选股池、结果和矩阵快照。"""
self._strategy_pools.clear()
@@ -413,6 +418,12 @@ class MonitorRuleEngine:
rule.get("scope", "symbols"),
tuple(sorted(str(symbol) for symbol in rule.get("symbols", []))),
rule.get("sector"),
rule.get("sector_kind"),
tuple(sorted(str(target.get("key")) for target in rule.get("sector_targets", []))),
rule.get("sector_trigger"),
rule.get("direction"),
rule.get("threshold_pct"),
rule.get("window_minutes"),
)
def set_rules(self, rules: list[dict]) -> None:
@@ -450,6 +461,11 @@ class MonitorRuleEngine:
for key, value in list(self._strategy_signal_seen.items())
if key[0] in active_ids
}
self._sector_condition_state = {
key: value
for key, value in list(self._sector_condition_state.items())
if key[0] in active_ids
}
logger.info("MonitorRuleEngine: 装载 %d 条规则", len(self._rules))
def add_rule(self, rule: dict) -> None:
@@ -470,6 +486,9 @@ class MonitorRuleEngine:
self._strategy_signal_seen = {
k: v for k, v in list(self._strategy_signal_seen.items()) if k[0] != rule_id
}
self._sector_condition_state = {
k: v for k, v in self._sector_condition_state.items() if k[0] != rule_id
}
def clear(self) -> None:
self._rules.clear()
@@ -477,6 +496,7 @@ class MonitorRuleEngine:
self._strategy_pools.clear()
self._strategy_signal_state.clear()
self._strategy_signal_seen.clear()
self._sector_condition_state.clear()
@property
def rules(self) -> dict[str, dict]:
@@ -640,6 +660,8 @@ class MonitorRuleEngine:
for rule_id, rule in list(self._rules.items()):
if rule.get("asset_type", "stock") != asset_type:
continue
if rule.get("type") == "sector":
continue
try:
events.extend(self._evaluate_rule(df, rule, now))
except Exception as e:
@@ -652,6 +674,156 @@ class MonitorRuleEngine:
return events
def evaluate_sectors(
self,
stock_df: pl.DataFrame,
index_df: pl.DataFrame,
*,
now: float | None = None,
) -> list[dict]:
"""按板块聚合快照评估 type=sector 规则。"""
if self._sector_monitor_service is None:
return []
rules = [
rule for rule in list(self._rules.values())
if rule.get("enabled", True) and rule.get("type") == "sector"
]
if not rules:
return []
targets_by_key: dict[str, dict] = {}
windows: set[int] = set()
for rule in rules:
for target in rule.get("sector_targets", []):
if target.get("key"):
targets_by_key[str(target["key"])] = target
if rule.get("sector_trigger") == "momentum":
windows.add(int(rule.get("window_minutes", 5)))
timestamp = time.time() if now is None else now
snapshots = self._sector_monitor_service.build_snapshots(
stock_df,
index_df,
list(targets_by_key.values()),
windows,
now=timestamp,
)
events: list[dict] = []
for rule in rules:
try:
events.extend(self._evaluate_sector_rule(rule, snapshots, timestamp))
except Exception as exc: # noqa: BLE001
logger.warning("板块规则评估失败 %s: %s", rule.get("id"), exc)
return events
def _evaluate_sector_rule(self, rule: dict, snapshots: dict[str, dict], now: float) -> list[dict]:
events: list[dict] = []
direction = rule.get("direction", "up")
trigger = rule.get("sector_trigger", "change_pct")
threshold = float(rule.get("threshold_pct", 1.0)) / 100
window = int(rule.get("window_minutes", 5))
for target in rule.get("sector_targets", []):
target_key = str(target.get("key") or "")
snapshot = snapshots.get(target_key)
if not snapshot or not snapshot.get("valid"):
continue
value = (
snapshot.get("change_pct")
if trigger == "change_pct"
else snapshot.get("window_changes", {}).get(window)
)
condition = value is not None and (
value >= threshold if direction == "up" else value <= -threshold
)
state_key = (rule["id"], target_key)
previous = self._sector_condition_state.get(state_key)
self._sector_condition_state[state_key] = condition
if previous is None or previous or not condition:
continue
event_type = f"sector_{trigger}_{direction}"
cooldown_key = (rule["id"], target_key, event_type)
last = self._last_fire.get(cooldown_key)
cooldown = int(rule.get("cooldown_seconds", 3600))
if last is not None and now - last < cooldown:
continue
self._last_fire[cooldown_key] = now
message = rule.get("message", "") or self._sector_message(
snapshot, trigger, direction, threshold, window, value,
)
event = {
"ts": int(now * 1000),
"rule_id": rule["id"],
"rule_name": rule.get("name", ""),
"strategy_id": None,
"source": "sector",
"type": event_type,
"symbol": snapshot.get("symbol") if snapshot.get("kind") == "index" else "",
"name": snapshot.get("name"),
"message": message,
"price": snapshot.get("price"),
"change_pct": snapshot.get("change_pct"),
"window_change_pct": value if trigger == "momentum" else None,
"signals": [],
"severity": rule.get("severity", "info"),
"conditions": [],
"logic": "and",
"sector_kind": snapshot.get("kind"),
"sector_key": target_key,
"sector_name": snapshot.get("name"),
"sector_source_field": snapshot.get("source_field"),
"sector_value": snapshot.get("value"),
"sector_level": snapshot.get("level"),
"coverage_ratio": snapshot.get("coverage_ratio"),
"valid_count": snapshot.get("valid_count"),
"total_count": snapshot.get("total_count"),
"up_count": snapshot.get("up_count"),
"down_count": snapshot.get("down_count"),
"leader": snapshot.get("leader"),
}
events.append(event)
if self._alert_handler:
try:
self._alert_handler(event)
except Exception as exc: # noqa: BLE001
logger.warning("alert handler failed: %s", exc)
return events
@staticmethod
def _sector_message(
snapshot: dict,
trigger: str,
direction: str,
threshold: float,
window: int,
value: float | None,
) -> str:
kind_label = {
"index": "指数", "concept": "概念", "industry": "行业",
}.get(snapshot.get("kind"), "板块")
current = float(snapshot.get("change_pct") or 0)
if trigger == "momentum":
action = "快速拉升" if direction == "up" else "快速下跌"
head = (
f"{kind_label}{snapshot.get('name')}{window}分钟{action} "
f"{float(value or 0) * 100:+.2f}%"
)
else:
action = "涨幅上穿" if direction == "up" else "跌幅下穿"
head = f"{kind_label}{snapshot.get('name')}{action} {threshold * 100:.2f}%"
parts = [head, f"当前 {current * 100:+.2f}%"]
if snapshot.get("kind") != "index":
parts.append(f"上涨 {snapshot.get('up_count', 0)}/{snapshot.get('valid_count', 0)}")
parts.append(f"覆盖 {float(snapshot.get('coverage_ratio') or 0) * 100:.0f}%")
leader = snapshot.get("leader") or {}
if leader.get("name") or leader.get("symbol"):
parts.append(
f"领涨 {leader.get('name') or leader.get('symbol')} "
f"{float(leader.get('change_pct') or 0) * 100:+.2f}%"
)
return "".join(parts)
def _evaluate_rule(self, df: pl.DataFrame, rule: dict, now: float) -> list[dict]:
"""评估单条规则,返回触发的 events。"""
# 1. 按 scope 过滤作用域
+37 -3
View File
@@ -27,7 +27,7 @@ logger = logging.getLogger(__name__)
# ── 常量 ────────────────────────────────────────────────
ID_RE = re.compile(r"^[a-z0-9_]{1,40}$")
RULE_TYPES = {"strategy", "signal", "price", "market", "ladder"}
RULE_TYPES = {"strategy", "signal", "price", "market", "ladder", "sector"}
SCOPES = {"symbols", "all", "sector"}
LOGICS = {"and", "or"}
DIRECTIONS = {"entry", "exit", "both"}
@@ -38,6 +38,9 @@ OPS = {">", ">=", "<", "<=", "==", "!="}
LADDER_METRICS = {"sealed_vol", "sealed_amount"}
# ladder 规则: 方向 (up=涨停炸板预警, down=跌停翘板预警)
LADDER_DIRECTIONS = {"up", "down"}
SECTOR_KINDS = {"index", "concept", "industry"}
SECTOR_TRIGGERS = {"change_pct", "momentum"}
SECTOR_WINDOWS = {1, 3, 5, 10, 15}
# 布尔信号列前缀 (op=truth 时 field 取这些)
_SIGNAL_PREFIXES = ("signal_", "csg_")
@@ -138,6 +141,29 @@ def validate(rule: dict) -> None:
thr = rule.get("threshold")
if not isinstance(thr, (int, float)) or thr < 0:
raise ValueError("threshold 必须是非负数字 (封单 ≤ 此值时报警)")
elif rule.get("type") == "sector":
kind = rule.get("sector_kind")
if kind not in SECTOR_KINDS:
raise ValueError(f"sector_kind 必须是 {SECTOR_KINDS} 之一")
targets = rule.get("sector_targets")
if not isinstance(targets, list) or not targets:
raise ValueError("板块监控至少选择一个监控对象")
if len(targets) > 20:
raise ValueError("板块监控对象最多 20 个")
for target in targets:
if not isinstance(target, dict) or not target.get("key") or not target.get("name"):
raise ValueError("板块监控对象格式错误")
if target.get("kind") != kind:
raise ValueError("板块监控对象类型必须一致")
if rule.get("sector_trigger") not in SECTOR_TRIGGERS:
raise ValueError(f"sector_trigger 必须是 {SECTOR_TRIGGERS} 之一")
if rule.get("direction") not in LADDER_DIRECTIONS:
raise ValueError("板块监控 direction 必须是 up 或 down")
threshold_pct = rule.get("threshold_pct")
if not isinstance(threshold_pct, (int, float)) or not 0 < threshold_pct <= 20:
raise ValueError("板块监控阈值必须大于 0 且不超过 20%")
if rule.get("sector_trigger") == "momentum" and rule.get("window_minutes") not in SECTOR_WINDOWS:
raise ValueError(f"板块异动窗口必须是 {sorted(SECTOR_WINDOWS)} 分钟之一")
else:
# 信号/价格/市场类型: 需要 conditions
conds = rule.get("conditions")
@@ -196,9 +222,14 @@ def normalize(rule: dict) -> dict:
r.setdefault("scope", "symbols")
r.setdefault("symbols", [])
r.setdefault("sector", None)
r.setdefault("sector_kind", None)
r.setdefault("sector_targets", [])
r.setdefault("sector_trigger", "change_pct")
r.setdefault("threshold_pct", 1.0)
r.setdefault("window_minutes", 5)
r.setdefault("strategy_id", None)
# direction 默认值: ladder 用 "up", 其余用 "entry"
r.setdefault("direction", "up" if r.get("type") == "ladder" else "entry")
# direction 默认值: ladder/sector 用 "up", 其余用 "entry"
r.setdefault("direction", "up" if r.get("type") in {"ladder", "sector"} else "entry")
if r.get("type") == "strategy":
if r.get("notify_events") is None:
# 兼容统一监控上线后的旧规则: 当时实际行为是同时通知进入和移出。
@@ -211,6 +242,9 @@ def normalize(rule: dict) -> dict:
# ladder 专属默认字段
r.setdefault("metric", "sealed_vol")
r.setdefault("threshold", 0)
if r.get("type") == "sector":
r["scope"] = "all"
r["symbols"] = []
r.setdefault("logic", "and")
r.setdefault("cooldown_seconds", 3600)
r.setdefault("severity", "info")
+216
View File
@@ -0,0 +1,216 @@
from __future__ import annotations
from types import SimpleNamespace
import polars as pl
import pytest
from app.services import sector_monitor
from app.services.ext_data import ExtConfig, ExtConfigStore, ExtField
from app.services.sector_monitor import SectorMonitorService
from app.strategy import monitor_rules
from app.strategy.monitor import MonitorRuleEngine
class _Repo:
def __init__(self, data_dir, indices: pl.DataFrame | None = None):
self.store = SimpleNamespace(data_dir=data_dir)
self._indices = indices if indices is not None else pl.DataFrame()
def get_index_instruments(self) -> pl.DataFrame:
return self._indices
def _index_target(symbol: str, name: str) -> dict:
return {
"key": f"index:{symbol}",
"kind": "index",
"name": name,
"symbol": symbol,
}
def _sector_rule(targets: list[dict], **overrides) -> dict:
rule = {
"id": "r_sector",
"name": "板块监控",
"enabled": True,
"type": "sector",
"scope": "all",
"sector_kind": targets[0]["kind"],
"sector_targets": targets,
"sector_trigger": "change_pct",
"direction": "up",
"threshold_pct": 1.0,
"window_minutes": 5,
"cooldown_seconds": 0,
"severity": "info",
}
rule.update(overrides)
return monitor_rules.normalize(rule)
def test_validate_accepts_sector_rule_and_rejects_mixed_target_kinds():
rule = _sector_rule([_index_target("000001.SH", "上证指数")])
monitor_rules.validate(rule)
mixed = _sector_rule([
_index_target("000001.SH", "上证指数"),
{
"key": "concept:test:field:人工智能",
"kind": "concept",
"name": "人工智能",
"source_id": "test",
"field": "field",
"value": "人工智能",
},
])
try:
monitor_rules.validate(mixed)
except ValueError as exc:
assert "类型" in str(exc)
else:
raise AssertionError("混合板块类型必须被拒绝")
def test_dimension_values_preserve_names_with_spaces_and_filter_nulls(tmp_path):
service = SectorMonitorService(_Repo(tmp_path))
assert service._dimension_values("中国AI 50;6G概念") == ["中国AI 50", "6G概念"]
assert service._dimension_values("nan") == []
assert service._dimension_values(float("nan")) == []
assert service._industry_paths("电子-半导体-数字芯片设计")[-1] == (
"电子-半导体-数字芯片设计", 3, "电子 / 半导体 / 数字芯片设计",
)
def test_index_targets_are_evaluated_independently(tmp_path):
repo = _Repo(tmp_path)
service = SectorMonitorService(repo)
engine = MonitorRuleEngine()
engine.set_sector_monitor_service(service)
sh = _index_target("000001.SH", "上证指数")
cyb = _index_target("399006.SZ", "创业板指")
engine.set_rules([_sector_rule([sh, cyb])])
first = pl.DataFrame({
"symbol": ["000001.SH", "399006.SZ"],
"name": ["上证指数", "创业板指"],
"close": [3000.0, 2000.0],
"change_pct": [0.8, 0.9],
})
assert engine.evaluate_sectors(pl.DataFrame(), first, now=1000.0) == []
second = first.with_columns(
pl.Series("change_pct", [1.2, 0.95]),
)
events = engine.evaluate_sectors(pl.DataFrame(), second, now=1006.0)
assert [event["sector_name"] for event in events] == ["上证指数"]
assert events[0]["change_pct"] == 0.012
def test_index_availability_updates_when_realtime_pool_changes(tmp_path, monkeypatch):
selected = ["000001.SH"]
monkeypatch.setattr(sector_monitor.preferences, "get_realtime_pull_index", lambda: True)
monkeypatch.setattr(sector_monitor.preferences, "get_realtime_index_mode", lambda: "core")
monkeypatch.setattr(sector_monitor.preferences, "get_realtime_index_symbols", lambda: selected)
service = SectorMonitorService(_Repo(tmp_path))
first = {target["symbol"]: target for target in service.list_targets()["index"]}
assert first["000001.SH"]["available"] is True
assert first["399006.SZ"]["available"] is False
initial_quote = pl.DataFrame({"symbol": ["000001.SH"], "change_pct": [0.2]})
service.build_snapshots(pl.DataFrame(), initial_quote, [first["000001.SH"]], {5}, now=1000.0)
selected[:] = ["399006.SZ"]
second = {target["symbol"]: target for target in service.list_targets()["index"]}
assert second["000001.SH"]["available"] is False
assert second["399006.SZ"]["available"] is True
changed_quote = pl.DataFrame({"symbol": ["000001.SH"], "change_pct": [1.3]})
snapshot = service.build_snapshots(
pl.DataFrame(), changed_quote, [second["000001.SH"]], {5}, now=1300.0,
)
assert snapshot["index:000001.SH"]["window_changes"][5] is None
def test_concept_snapshot_uses_member_average_and_full_window(tmp_path):
config = ExtConfig(
id="concept_test",
label="概念测试",
mode="snapshot",
fields=[
ExtField("symbol", "string", "标的代码"),
ExtField("concept", "string", "所属概念"),
],
)
ExtConfigStore(tmp_path).upsert(config)
ext_dir = tmp_path / "ext_data" / config.id
pl.DataFrame({
"symbol": ["A", "B", "C", "D", "E"],
"concept": ["人工智能", "人工智能", "人工智能", "人工智能", "人工智能"],
}).write_parquet(ext_dir / "part.parquet")
service = SectorMonitorService(_Repo(tmp_path))
target = next(
target for target in service.list_targets()["concept"]
if target["name"] == "人工智能"
)
first = pl.DataFrame({
"symbol": ["A", "B", "C", "D", "E"],
"name": ["", "", "", "", ""],
"close": [10.0] * 5,
"change_pct": [0.01, 0.02, 0.03, -0.01, 0.0],
})
snapshots = service.build_snapshots(first, pl.DataFrame(), [target], {5}, now=1000.0)
assert snapshots[target["key"]]["change_pct"] == pytest.approx(0.01)
assert snapshots[target["key"]]["coverage_ratio"] == 1.0
assert snapshots[target["key"]]["window_changes"][5] is None
second = first.with_columns((pl.col("change_pct") + 0.01).alias("change_pct"))
too_early = service.build_snapshots(second, pl.DataFrame(), [target], {5}, now=1240.0)
assert too_early[target["key"]]["window_changes"][5] is None
unrelated = ExtConfig(
id="hot_test",
label="热度测试",
mode="snapshot",
fields=[
ExtField("symbol", "string", "标的代码"),
ExtField("heat", "float", "市场热度"),
],
)
ExtConfigStore(tmp_path).upsert(unrelated)
unrelated_dir = tmp_path / "ext_data" / unrelated.id
pl.DataFrame({"symbol": ["A"], "heat": [1.0]}).write_parquet(unrelated_dir / "part.parquet")
complete = service.build_snapshots(second, pl.DataFrame(), [target], {5}, now=1300.0)
assert complete[target["key"]]["window_changes"][5] == pytest.approx(0.01)
def test_momentum_rule_triggers_after_complete_window(tmp_path):
service = SectorMonitorService(_Repo(tmp_path))
engine = MonitorRuleEngine()
engine.set_sector_monitor_service(service)
target = _index_target("000001.SH", "上证指数")
engine.set_rules([_sector_rule(
[target],
sector_trigger="momentum",
threshold_pct=1.0,
window_minutes=5,
)])
start = pl.DataFrame({
"symbol": ["000001.SH"],
"name": ["上证指数"],
"close": [3000.0],
"change_pct": [0.2],
})
assert engine.evaluate_sectors(pl.DataFrame(), start, now=1000.0) == []
early = start.with_columns(pl.lit(1.3).alias("change_pct"))
assert engine.evaluate_sectors(pl.DataFrame(), early, now=1240.0) == []
events = engine.evaluate_sectors(pl.DataFrame(), early, now=1300.0)
assert len(events) == 1
assert events[0]["type"] == "sector_momentum_up"
assert events[0]["window_change_pct"] == pytest.approx(0.011)
+1
View File
@@ -88,6 +88,7 @@ const SOURCE_BADGE: Record<string, { label: string; cls: string }> = {
signal: { label: '信号', cls: 'bg-accent/15 text-accent' },
price: { label: '价格', cls: 'bg-emerald-400/15 text-emerald-400' },
market: { label: '异动', cls: 'bg-purple-500/15 text-purple-400' },
sector: { label: '板块', cls: 'bg-cyan-500/15 text-cyan-700 dark:text-cyan-300' },
pool_entry: { label: '进入', cls: 'bg-emerald-400/15 text-emerald-400' },
pool_exit: { label: '移出', cls: 'bg-warning/15 text-warning' },
buy_signal: { label: '买入', cls: 'bg-danger/15 text-danger' },
+261 -11
View File
@@ -1,8 +1,8 @@
import { useState } from 'react'
import { Link } from 'react-router-dom'
import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query'
import { Activity, Check, Plus, RadioTower, Save, Search, TrendingUp, Waypoints, X } from 'lucide-react'
import { api, genRuleId, type MonitorRule, type MonitorCondition, type StrategyNotifyEvent } from '@/lib/api'
import { Activity, Building2, ChartNoAxesCombined, Check, Layers3, Plus, RadioTower, Save, Search, Tags, TrendingUp, Waypoints, X } from 'lucide-react'
import { api, genRuleId, type MonitorRule, type MonitorCondition, type SectorKind, type SectorMonitorTarget, type StrategyNotifyEvent } from '@/lib/api'
import { DEFAULT_STRATEGY_NOTIFY_EVENTS, LEGACY_STRATEGY_NOTIFY_EVENTS, STRATEGY_NOTIFY_EVENT_OPTIONS } from '@/lib/strategyMonitorEvents'
import { QK } from '@/lib/queryKeys'
import { boardTag } from '@/components/stock-table/primitives'
@@ -22,7 +22,7 @@ interface Props {
}
const TYPE_DEFAULT_NAME: Record<string, string> = {
signal: '信号监控', price: '价格监控', market: '市场异动监控', strategy: '策略监控',
signal: '信号监控', price: '价格监控', market: '市场异动监控', strategy: '策略监控', sector: '板块监控',
}
const TYPE_ICONS = {
@@ -30,8 +30,15 @@ const TYPE_ICONS = {
price: TrendingUp,
market: RadioTower,
strategy: Waypoints,
sector: Layers3,
}
const SECTOR_KIND_OPTIONS: Array<{ key: SectorKind; label: string; icon: typeof ChartNoAxesCombined }> = [
{ key: 'index', label: '大盘指数', icon: ChartNoAxesCombined },
{ key: 'concept', label: '概念题材', icon: Tags },
{ key: 'industry', label: '行业板块', icon: Building2 },
]
const STRATEGY_SOURCE_META = {
builtin: { label: '内置', className: 'border-accent/25 bg-accent/10 text-accent' },
custom: { label: '自定义', className: 'border-emerald-400/25 bg-emerald-400/10 text-emerald-400' },
@@ -48,6 +55,11 @@ const emptyRule = (preset?: Partial<MonitorRule>): MonitorRule => ({
scope: 'symbols',
symbols: [],
sector: null,
sector_kind: 'index',
sector_targets: [],
sector_trigger: 'change_pct',
threshold_pct: 1,
window_minutes: 5,
strategy_id: null,
direction: 'entry',
conditions: [],
@@ -75,6 +87,7 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) {
? [...(rule.notify_events ?? LEGACY_STRATEGY_NOTIFY_EVENTS)]
: undefined,
conditions: rule.conditions.map(c => ({ ...c })),
sector_targets: rule.sector_targets?.map(target => ({ ...target })) ?? [],
}
}
const initial = {
@@ -94,6 +107,11 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) {
})
const [error, setError] = useState('')
const [symbolQuery, setSymbolQuery] = useState('')
const [sectorQuery, setSectorQuery] = useState('')
const [industryLevel, setIndustryLevel] = useState<1 | 2 | 3>(() => {
const level = rule?.sector_targets?.[0]?.level
return level === 1 || level === 3 ? level : 2
})
const [strategyQuery, setStrategyQuery] = useState('')
const [strategyCategory, setStrategyCategory] = useState<'all' | 'builtin' | 'custom' | 'ai' | 'composite'>('all')
// 标的搜索资产类型: ETF 一并搜股票; 指数只搜指数; 否则只搜股票。
@@ -111,13 +129,22 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) {
// name 为空时用默认名
if (!d.name.trim()) {
const base = TYPE_DEFAULT_NAME[d.type] ?? '监控规则'
d.name = d.scope === 'symbols' && d.symbols.length > 0
d.name = d.type === 'sector' && d.sector_targets?.length
? `${base} · ${d.sector_targets[0].name}${d.sector_targets.length > 1 ? `${d.sector_targets.length}` : ''}`
: d.scope === 'symbols' && d.symbols.length > 0
? `${base} · ${d.symbols[0]}${d.symbols.length > 1 ? `${d.symbols.length}` : ''}`
: base
}
if (d.type === 'strategy') {
if (!d.strategy_id) throw new Error('策略监控必须选择一个策略')
if (!d.notify_events?.length) throw new Error('至少选择一个通知事件')
} else if (d.type === 'sector') {
d.scope = 'all'
d.symbols = []
d.conditions = []
delete d.notify_events
if (!d.sector_targets?.length) throw new Error('请选择至少一个监控对象')
if ((d.threshold_pct ?? 0) <= 0 || (d.threshold_pct ?? 0) > 20) throw new Error('阈值必须大于 0 且不超过 20%')
} else {
delete d.notify_events
if (d.conditions.length === 0) throw new Error('至少选择一个触发条件')
@@ -126,7 +153,7 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) {
if (c.op !== 'truth' && (c.value === null || c.value === undefined)) throw new Error('阈值条件需要数值')
}
}
if (d.scope === 'symbols' && d.symbols.length === 0) throw new Error('请选择至少一只标的')
if (d.type !== 'sector' && d.scope === 'symbols' && d.symbols.length === 0) throw new Error('请选择至少一只标的')
return api.monitorRuleSave(d)
},
onSuccess: () => {
@@ -158,6 +185,22 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) {
setSymbolQuery('')
}
const selectSectorKind = (kind: SectorKind) => {
setDraft(d => ({ ...d, sector_kind: kind, sector_targets: [] }))
setSectorQuery('')
}
const toggleSectorTarget = (target: SectorMonitorTarget) => {
setDraft(d => {
const current = d.sector_targets ?? []
if (current.some(item => item.key === target.key)) {
return { ...d, sector_targets: current.filter(item => item.key !== target.key) }
}
if (current.length >= 20) return d
return { ...d, sector_targets: [...current, target] }
})
}
// 勾选/取消勾选某个推送渠道 (飞书 / 企业微信 各自独立)
const toggleChannel = (ch: string) =>
setDraft(d => {
@@ -199,6 +242,14 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) {
const visibleScopes = (options.data?.scopes ?? []).filter(
s => assetType !== 'index' || s.key === 'symbols',
)
const sectorKind = draft.sector_kind ?? 'index'
const sectorTargets = options.data?.sector_targets?.[sectorKind] ?? []
const visibleSectorTargets = sectorTargets.filter(target => {
if (sectorKind === 'industry' && target.level !== industryLevel) return false
const query = sectorQuery.trim().toLowerCase()
if (!query) return true
return `${target.name} ${target.symbol ?? ''} ${target.value ?? ''}`.toLowerCase().includes(query)
}).slice(0, 100)
const thresholdConds = draft.conditions.filter(c => c.op !== 'truth')
const strategyPresets = strategies.data?.presets ?? []
const normalizedStrategyQuery = strategyQuery.trim().toLowerCase()
@@ -329,7 +380,7 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) {
</div>
{/* 资产类型: 股票 / ETF / 指数 (个股极简模式不显示) */}
{!simple && (
{!simple && draft.type !== 'sector' && (
<div className="space-y-1.5">
<span className="text-[11px] text-muted"></span>
<div className="inline-flex h-9 rounded-btn border border-border overflow-hidden">
@@ -364,7 +415,7 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) {
{/* 监控类型 */}
<div className="space-y-1.5">
<span className="text-[11px] text-muted"></span>
<div className="grid grid-cols-2 gap-1.5 sm:grid-cols-4">
<div className="grid grid-cols-2 gap-1.5 sm:grid-cols-5">
{visibleTypes.map(t => {
const Icon = TYPE_ICONS[t.key as keyof typeof TYPE_ICONS] ?? Activity
const active = draft.type === t.key
@@ -381,7 +432,10 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) {
notify_events: type === 'strategy'
? [...(d.notify_events ?? DEFAULT_STRATEGY_NOTIFY_EVENTS)]
: undefined,
scope: type === 'strategy' && d.scope === 'symbols' && d.symbols.length === 0 ? 'all' : d.scope,
scope: type === 'sector'
? 'all'
: type === 'strategy' && d.scope === 'symbols' && d.symbols.length === 0 ? 'all' : d.scope,
direction: type === 'sector' ? 'up' : d.type === 'sector' ? 'entry' : d.direction,
}
})}
className={`inline-flex h-9 items-center justify-center gap-1.5 rounded-btn border px-2 text-xs font-medium transition-colors cursor-pointer ${
@@ -403,8 +457,204 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) {
<input value={draft.name} onChange={e => setDraft(d => ({ ...d, name: e.target.value }))} placeholder="留空用默认名称" className="h-9 w-full rounded-btn border border-border bg-base px-3 text-xs text-foreground" />
</label>
{draft.type === 'sector' && (
<div className="space-y-4 border-t border-border/60 pt-4">
<div className="space-y-1.5">
<span className="text-[11px] text-muted"></span>
<div className="grid grid-cols-3 gap-1.5">
{SECTOR_KIND_OPTIONS.map(option => {
const Icon = option.icon
const active = sectorKind === option.key
return (
<button
key={option.key}
type="button"
aria-pressed={active}
onClick={() => selectSectorKind(option.key)}
className={`inline-flex h-9 items-center justify-center gap-1.5 rounded-btn border text-xs font-medium transition-colors cursor-pointer ${
active ? 'border-accent/40 bg-accent/10 text-accent' : 'border-border bg-base text-secondary hover:border-accent/25'
}`}
>
<Icon className="h-3.5 w-3.5" />
{option.label}
</button>
)
})}
</div>
</div>
{sectorKind === 'industry' && (
<div className="space-y-1.5">
<span className="text-[11px] text-muted"></span>
<div className="inline-flex h-8 overflow-hidden rounded-btn border border-border bg-base">
{([1, 2, 3] as const).map(level => (
<button
key={level}
type="button"
aria-pressed={industryLevel === level}
onClick={() => {
setIndustryLevel(level)
setDraft(d => ({ ...d, sector_targets: [] }))
}}
className={`px-3 text-[11px] transition-colors cursor-pointer ${
industryLevel === level ? 'bg-accent/10 text-accent' : 'text-muted hover:text-foreground'
}`}
>
{level}
</button>
))}
</div>
</div>
)}
<div className="space-y-2">
<div className="flex items-center justify-between gap-3">
<span className="text-[11px] text-muted"></span>
<span className="text-[10px] font-mono text-muted">{draft.sector_targets?.length ?? 0}/20</span>
</div>
{(draft.sector_targets?.length ?? 0) > 0 && (
<div className="flex flex-wrap gap-1">
{draft.sector_targets?.map(target => (
<span key={target.key} className="inline-flex items-center gap-1 rounded bg-accent/8 px-1.5 py-1 text-[10px] text-accent">
{target.name}
<button type="button" onClick={() => toggleSectorTarget(target)} title="移除" className="text-accent/60 hover:text-danger cursor-pointer">
<X className="h-2.5 w-2.5" />
</button>
</span>
))}
</div>
)}
<label className="relative block">
<Search className="absolute left-2.5 top-2.5 h-3.5 w-3.5 text-muted" />
<input
value={sectorQuery}
onChange={event => setSectorQuery(event.target.value)}
placeholder={`搜索${SECTOR_KIND_OPTIONS.find(option => option.key === sectorKind)?.label ?? '板块'}`}
className="h-9 w-full rounded-btn border border-border bg-base pl-8 pr-3 text-xs text-foreground placeholder:text-muted/50 focus:border-accent/50 focus:outline-none"
/>
</label>
<div className="grid max-h-48 grid-cols-1 gap-1 overflow-y-auto pr-1 sm:grid-cols-2">
{visibleSectorTargets.length === 0 ? (
<div className="col-span-full rounded-btn border border-dashed border-border py-6 text-center text-xs text-muted">
{options.isLoading ? '正在加载...' : '没有可用的监控对象'}
</div>
) : visibleSectorTargets.map(target => {
const selected = draft.sector_targets?.some(item => item.key === target.key) ?? false
const unavailable = !target.available || (target.kind !== 'index' && target.member_count < 5)
const targetLabel = target.kind === 'industry'
? (target.value ?? target.name).replaceAll('-', ' / ')
: target.name
return (
<button
key={target.key}
type="button"
disabled={unavailable}
aria-pressed={selected}
onClick={() => toggleSectorTarget(target)}
title={!target.available ? '请先在实时监控设置中加入该指数' : target.member_count < 5 ? '有效成分少于 5 只' : targetLabel}
className={`flex h-9 min-w-0 items-center gap-2 rounded-btn border px-2.5 text-left transition-colors ${
unavailable
? 'cursor-not-allowed border-border/40 bg-base/40 text-muted/40'
: selected
? 'cursor-pointer border-accent/40 bg-accent/10 text-accent'
: 'cursor-pointer border-border bg-base text-secondary hover:border-accent/25 hover:text-foreground'
}`}
>
<span className="min-w-0 flex-1 truncate text-[11px]">{targetLabel}</span>
{target.symbol && <span className="shrink-0 font-mono text-[9px] opacity-60">{target.symbol}</span>}
{target.kind !== 'index' && <span className="shrink-0 font-mono text-[9px] opacity-60">{target.member_count}</span>}
<span className={`grid h-4 w-4 shrink-0 place-items-center rounded-full border ${selected ? 'border-accent bg-accent text-white' : 'border-border text-transparent'}`}>
<Check className="h-2.5 w-2.5" />
</span>
</button>
)
})}
</div>
</div>
<div className="grid gap-3 border-t border-border/60 pt-4 sm:grid-cols-2">
<div className="space-y-1.5">
<span className="text-[11px] text-muted"></span>
<div className="grid h-9 grid-cols-2 overflow-hidden rounded-btn border border-border bg-base">
{([
['change_pct', '涨跌幅到达'],
['momentum', '快速异动'],
] as const).map(([key, label]) => (
<button
key={key}
type="button"
aria-pressed={(draft.sector_trigger ?? 'change_pct') === key}
onClick={() => setDraft(d => ({ ...d, sector_trigger: key }))}
className={`text-[11px] font-medium transition-colors cursor-pointer ${
(draft.sector_trigger ?? 'change_pct') === key ? 'bg-accent/10 text-accent' : 'text-muted hover:text-foreground'
}`}
>
{label}
</button>
))}
</div>
</div>
<div className="space-y-1.5">
<span className="text-[11px] text-muted"></span>
<div className="grid h-9 grid-cols-2 overflow-hidden rounded-btn border border-border bg-base">
{([
['up', draft.sector_trigger === 'momentum' ? '快速上涨' : '上涨'],
['down', draft.sector_trigger === 'momentum' ? '快速下跌' : '下跌'],
] as const).map(([key, label]) => (
<button
key={key}
type="button"
aria-pressed={draft.direction === key}
onClick={() => setDraft(d => ({ ...d, direction: key }))}
className={`text-[11px] font-medium transition-colors cursor-pointer ${
draft.direction === key ? 'bg-accent/10 text-accent' : 'text-muted hover:text-foreground'
}`}
>
{label}
</button>
))}
</div>
</div>
{draft.sector_trigger === 'momentum' && (
<label className="space-y-1.5">
<span className="text-[11px] text-muted"></span>
<select
value={draft.window_minutes ?? 5}
onChange={event => setDraft(d => ({ ...d, window_minutes: Number(event.target.value) as MonitorRule['window_minutes'] }))}
className="h-9 w-full rounded-btn border border-border bg-base px-3 text-xs text-foreground"
>
{[1, 3, 5, 10, 15].map(window => <option key={window} value={window}>{window} </option>)}
</select>
</label>
)}
<label className="space-y-1.5">
<span className="text-[11px] text-muted">{draft.sector_trigger === 'momentum' ? '窗口变化阈值' : '板块涨跌幅阈值'}</span>
<span className="relative block">
<input
type="number"
min="0.01"
max="20"
step="0.1"
value={draft.threshold_pct ?? 1}
onChange={event => setDraft(d => ({ ...d, threshold_pct: Number(event.target.value) }))}
className="h-9 w-full rounded-btn border border-border bg-base pl-3 pr-8 text-xs font-mono text-foreground"
/>
<span className="absolute right-3 top-2.5 text-xs text-muted">%</span>
</span>
</label>
</div>
{sectorKind !== 'index' && (
<div className="flex flex-wrap gap-1.5 text-[9px] text-muted">
<span className="rounded bg-elevated px-1.5 py-0.5"></span>
<span className="rounded bg-elevated px-1.5 py-0.5"> 80%</span>
<span className="rounded bg-elevated px-1.5 py-0.5"> 5</span>
</div>
)}
</div>
)}
{/* 作用范围 */}
<div className="space-y-2">
{draft.type !== 'sector' && <div className="space-y-2">
<span className="text-[11px] text-muted"></span>
<div className="flex items-center gap-2">
<select value={draft.scope} onChange={e => setDraft(d => ({ ...d, scope: e.target.value as MonitorRule['scope'] }))} className="h-9 w-32 rounded-btn border border-border bg-base px-3 text-xs text-foreground">
@@ -445,10 +695,10 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) {
{draft.scope === 'all' && <span className="text-[11px] text-muted"></span>}
{draft.scope === 'sector' && <span className="text-[11px] text-muted/60">(,)</span>}
</div>
</div>
</div>}
{/* 触发条件 (非 strategy) */}
{draft.type !== 'strategy' && (
{draft.type !== 'strategy' && draft.type !== 'sector' && (
<div className="space-y-3">
<div className="flex items-center justify-between">
<span className="text-[11px] text-muted"></span>
+36 -1
View File
@@ -622,15 +622,36 @@ export interface MonitorCondition {
export type StrategyNotifyEvent = 'buy_signal' | 'sell_signal' | 'pool_entry' | 'pool_exit'
export type SectorKind = 'index' | 'concept' | 'industry'
export interface SectorMonitorTarget {
key: string
kind: SectorKind
name: string
symbol?: string
source_id?: string
field?: string
source_field?: string
value?: string
level?: number | null
available: boolean
member_count: number
}
export interface MonitorRule {
id: string
name: string
enabled: boolean
type: 'strategy' | 'signal' | 'price' | 'market' | 'ladder'
type: 'strategy' | 'signal' | 'price' | 'market' | 'ladder' | 'sector'
asset_type?: 'stock' | 'etf' | 'index'
scope: 'symbols' | 'all' | 'sector'
symbols: string[]
sector?: string | null
sector_kind?: SectorKind | null
sector_targets?: SectorMonitorTarget[]
sector_trigger?: 'change_pct' | 'momentum'
threshold_pct?: number
window_minutes?: 1 | 3 | 5 | 10 | 15
strategy_id?: string | null
direction: 'entry' | 'exit' | 'both' | 'up' | 'down'
notify_events?: StrategyNotifyEvent[]
@@ -665,6 +686,7 @@ export interface MonitorRuleOptions {
max_symbols: number
reason: string
}
sector_targets: Record<SectorKind, SectorMonitorTarget[]>
}
export interface AlertEvent {
@@ -683,6 +705,19 @@ export interface AlertEvent {
strategy_id?: string
conditions?: MonitorCondition[]
logic?: 'and' | 'or'
sector_kind?: SectorKind
sector_key?: string
sector_name?: string
sector_source_field?: string
sector_value?: string
sector_level?: number | null
window_change_pct?: number | null
coverage_ratio?: number
valid_count?: number
total_count?: number
up_count?: number
down_count?: number
leader?: { symbol?: string; name?: string; change_pct?: number } | null
/** ext 富化字段 (行业/概念等), 键为 "{configId}__{fieldName}" */
[key: string]: unknown
}
+5
View File
@@ -116,6 +116,11 @@ function buildSingleText(a: AlertEvent): string {
const name = a.name || '标的'
const pctText = a.change_pct != null ? fmtPctText(a.change_pct) : ''
// 板块消息已包含名称、触发条件和当前涨跌幅,避免重复播报。
if (a.source === 'sector') {
return a.message || name
}
// 策略类: message 存的是策略名(单条) 或完整批量描述(>5只)
if (a.source === 'strategy') {
// 批量事件 (symbol 为空/为 _batch): message 已含 "策略「X」进入 N 只:…" 直接念
+8 -5
View File
@@ -1,5 +1,5 @@
import { useState, useEffect, useRef, type ReactNode } from 'react'
import { Link } from 'react-router-dom'
import { Link, useNavigate } from 'react-router-dom'
import { useQuery, useMutation, useQueryClient } from '@tanstack/react-query'
import { motion, AnimatePresence } from 'framer-motion'
import { Activity, ArrowDownRight, ArrowUpRight, BarChart3, BellRing, Database, Flame, Gauge, Info, LineChart, Loader2, Play, RefreshCw, Sparkles, Target, Timer } from 'lucide-react'
@@ -87,15 +87,17 @@ const _SOURCE_BADGE: Record<string, string> = {
signal: 'bg-accent/10 text-accent',
price: 'bg-emerald-400/10 text-emerald-400',
market: 'bg-purple-500/10 text-purple-400',
sector: 'bg-cyan-500/10 text-cyan-700 dark:text-cyan-300',
}
const _SOURCE_LABEL: Record<string, string> = {
strategy: '策略', signal: '信号', price: '价格', market: '异动',
strategy: '策略', signal: '信号', price: '价格', market: '异动', sector: '板块',
}
const _SEVERITY_BAR: Record<string, string> = {
info: 'bg-accent/40', warn: 'bg-warning', critical: 'bg-danger',
}
function MonitorWidget({ onStockClick }: { onStockClick: (event: AlertEvent) => void }) {
const navigate = useNavigate()
const alerts = useQuery({
queryKey: ['alerts', ''],
queryFn: () => api.alertsList({ days: 7, limit: 10 }),
@@ -117,6 +119,7 @@ function MonitorWidget({ onStockClick }: { onStockClick: (event: AlertEvent) =>
const sev = _SEVERITY_BAR[ev.severity ?? 'info'] ?? _SEVERITY_BAR.info
const pct = ev.change_pct ?? 0
const isStrategy = ev.source === 'strategy'
const isSector = ev.source === 'sector'
const sname = isStrategy ? strategyName(ev.message ?? '') : ''
const eventMeta = strategyEventMeta(ev.type)
return (
@@ -131,9 +134,9 @@ function MonitorWidget({ onStockClick }: { onStockClick: (event: AlertEvent) =>
{/* 第一行: 代码 + 名称 + 价格 + 涨跌幅 (点击代码/名称弹日K) */}
<div className="flex items-center gap-1.5">
<button
onClick={() => ev.symbol && onStockClick(ev)}
title={ev.symbol ? `查看 ${ev.symbol} 日K` : undefined}
className="inline-flex items-center gap-1 min-w-0 shrink-0 rounded hover:bg-elevated/60 transition-colors -mx-0.5 px-0.5 cursor-pointer"
onClick={() => isSector ? navigate('/monitor') : ev.symbol && onStockClick(ev)}
title={isSector ? '在监控中心查看板块告警' : ev.symbol ? `查看 ${ev.symbol} 日K` : undefined}
className={`inline-flex items-center gap-1 min-w-0 shrink-0 rounded hover:bg-elevated/60 transition-colors -mx-0.5 px-0.5 ${isSector || ev.symbol ? 'cursor-pointer' : 'cursor-default'}`}
>
<span className="font-mono text-[10px] font-medium text-foreground/80 hover:text-accent">{ev.symbol?.replace(/\.(SH|SZ|BJ)$/, '')}</span>
{ev.symbol && (() => {
+46 -6
View File
@@ -1,4 +1,5 @@
import { useState, useRef, useEffect, useMemo } from 'react'
import { useNavigate } from 'react-router-dom'
import { useQuery, useMutation, useQueryClient } from '@tanstack/react-query'
import { motion, AnimatePresence } from 'framer-motion'
import { AlertTriangle, RadioTower, Plus, Trash2, Settings2, Zap, Bell, ListChecks, BellRing, TrendingUp, TrendingDown, Flame, Tags } from 'lucide-react'
@@ -20,7 +21,7 @@ import { DimensionMembersDialog, type DimensionKind, type DimensionMembersTarget
import { usePreferences } from '@/lib/useSharedQueries'
const TYPE_LABEL: Record<string, string> = {
signal: '信号', price: '价格/涨跌', market: '市场异动', strategy: '策略监控',
signal: '信号', price: '价格/涨跌', market: '市场异动', strategy: '策略监控', sector: '板块监控',
}
/** 严重级别 → 左侧色条 + 图标 */
@@ -34,6 +35,7 @@ const SOURCE_BADGE_STYLE: Record<string, string> = {
signal: 'bg-accent/10 text-accent border-accent/20',
price: 'bg-emerald-400/10 text-emerald-400 border-emerald-400/20',
market: 'bg-purple-500/10 text-purple-400 border-purple-500/20',
sector: 'bg-cyan-500/10 text-cyan-700 border-cyan-500/20 dark:text-cyan-300',
}
/**
@@ -114,7 +116,7 @@ export function Monitor() {
const [editingRule, setEditingRule] = useState<MonitorRule | null>(null)
// 触发记录: 过滤 + 统计 (提升到主组件, 供 header 行使用)
const [filter, setFilter] = useState<'all' | 'strategy' | 'signal' | 'price' | 'market'>('all')
const [filter, setFilter] = useState<'all' | 'strategy' | 'signal' | 'price' | 'market' | 'sector'>('all')
const [confirmClear, setConfirmClear] = useState(false)
const [confirmClearRules, setConfirmClearRules] = useState(false)
@@ -175,7 +177,7 @@ export function Monitor() {
<SectionHeader icon={BellRing} title="触发记录" />
{/* 过滤标签 */}
<div className="flex flex-wrap items-center gap-0.5">
{(['all', 'strategy', 'signal', 'price', 'market'] as const).map(f => (
{(['all', 'strategy', 'signal', 'price', 'market', 'sector'] as const).map(f => (
<button
key={f}
onClick={() => setFilter(f)}
@@ -294,6 +296,7 @@ function AlertsList({ alertsQuery, confirmClear, setConfirmClear, total, enterTs
monitorExtFields: { concept: MonitorExtFieldItem | null; industry: MonitorExtFieldItem | null }
}) {
const qc = useQueryClient()
const navigate = useNavigate()
const [confirmTs, setConfirmTs] = useState<number | null>(null)
const resetTimer = useRef<ReturnType<typeof setTimeout> | null>(null)
const [previewEv, setPreviewEv] = useState<AlertEvent | null>(null)
@@ -431,7 +434,28 @@ function AlertsList({ alertsQuery, confirmClear, setConfirmClear, total, enterTs
})() : (
<>
<div className="flex items-center gap-2 flex-wrap">
{ev.symbol && (() => {
{ev.source === 'sector' && (
<button
onClick={() => {
if (ev.sector_kind === 'index' && ev.symbol) {
navigate(`/indices?symbol=${encodeURIComponent(ev.symbol)}`)
} else if (ev.sector_source_field && ev.sector_value) {
setDimensionTarget({
kind: ev.sector_kind as DimensionKind,
value: ev.sector_value,
sourceField: ev.sector_source_field,
})
}
}}
className="inline-flex items-center gap-1.5 rounded px-1 -mx-1 text-xs font-medium text-foreground transition-colors hover:bg-elevated/50 hover:text-accent cursor-pointer"
title={ev.sector_kind === 'index' ? '打开指数详情' : '查看成分股'}
>
<Tags className="h-3.5 w-3.5 text-cyan-600 dark:text-cyan-300" />
<span>{ev.sector_name ?? ev.name}</span>
{ev.symbol && <span className="font-mono text-[10px] text-muted">{ev.symbol}</span>}
</button>
)}
{ev.symbol && ev.source !== 'sector' && (() => {
const board = boardTag(ev.symbol)
return (
<button
@@ -737,8 +761,24 @@ function RulesList({ rulesQuery, onEdit }: {
</div>
)}
{/* 第二行: 策略类型显示通知事件 */}
{r.type === 'strategy' && r.strategy_id ? (
{/* 第二行: 类型摘要 */}
{r.type === 'sector' ? (
<div className="mt-1 flex min-w-0 flex-wrap items-center gap-1 pl-0.5">
{(r.sector_targets ?? []).slice(0, 3).map(target => (
<span key={target.key} className="max-w-28 truncate rounded bg-cyan-500/8 px-1.5 py-0.5 text-[9px] text-cyan-700 dark:text-cyan-300">
{target.name}
</span>
))}
{(r.sector_targets?.length ?? 0) > 3 && (
<span className="text-[9px] text-muted">+{(r.sector_targets?.length ?? 0) - 3}</span>
)}
<span className="text-[9px] text-secondary">·</span>
<span className="text-[9px] text-secondary">
{r.sector_trigger === 'momentum' ? `${r.window_minutes ?? 5}分钟异动` : '涨跌幅'}
{r.direction === 'down' ? ' ≤ -' : ' ≥ '}{r.threshold_pct ?? 1}%
</span>
</div>
) : r.type === 'strategy' && r.strategy_id ? (
<div className="mt-1 flex flex-wrap items-center gap-1 pl-0.5">
{(r.notify_events ?? LEGACY_STRATEGY_NOTIFY_EVENTS).map(event => {
const option = STRATEGY_NOTIFY_EVENT_OPTIONS.find(item => item.key === event)
+3
View File
@@ -621,6 +621,7 @@ data/strategies/ai/
- **个股信号**:监控信号库中的技术或自定义信号。
- **价格/涨跌监控**:监控价格、涨跌幅或上下穿条件。
- **市场异动**:监控涨跌停、连板或其他市场状态。
- **板块监控**:监控大盘指数、概念题材或行业板块的涨跌幅和分钟异动。
### 15.2 创建规则
@@ -634,6 +635,8 @@ data/strategies/ai/
AND 要求所有条件同时成立;OR 表示任一条件成立即可。全市场规则计算范围大,创建前应确认数据权限和刷新频率。
板块监控中,一条大盘规则可选择多个指数,系统会对每个指数独立判断和触发。概念与行业按有效成分股等权计算,有效成分不少于 5 只且实时行情覆盖率达到 80% 时才会触发。分钟异动在收集到完整窗口后才开始判断,首份行情只用于建立基线。
### 15.3 内置信号示例
系统可提供日内分时价格上穿/下穿均价、指标上穿/下穿 0 轴等信号。涉及日内分时的信号需要: