diff --git a/backend/app/api/lots.py b/backend/app/api/lots.py new file mode 100644 index 0000000..739b30e --- /dev/null +++ b/backend/app/api/lots.py @@ -0,0 +1,126 @@ +"""批次登记 API — 薄"批次"页 (持仓提醒), 只做胶水, 不含会计语义。 + +映射/校验/持久化在 strategy.lots 域; 写完派生规则后复用 monitor_rules 的 _sync_engine 同步引擎。 +""" +from __future__ import annotations + +import secrets +import threading +import time +from pathlib import Path + +from fastapi import APIRouter, HTTPException, Request +from pydantic import BaseModel + +from app.strategy import lots as lots_domain +from app.strategy import monitor_rules + +router = APIRouter(prefix="/api/lots", tags=["lots"]) + +# 批次 + 派生规则 + 引擎重载的跨请求互斥; 规则全部校验通过才落盘, 避免半成品 (镜像 watchlist 服务层)。 +_write_lock = threading.Lock() + + +def _data_dir(request: Request) -> Path: + return request.app.state.repo.store.data_dir + + +def _resolve_asset_type(request: Request, symbol: str) -> str: + """按 symbol 解析资产类型 (stock/etf); 解析失败默认 stock (fail-safe)。""" + repo = getattr(request.app.state, "repo", None) + try: + return repo.resolve_asset_type(symbol) if repo is not None else "stock" + except Exception: + return "stock" + + +class LotModel(BaseModel): + id: str | None = None + symbol: str + qty: float = 0 + cost_price: float = 0 + buy_date: str | None = None + target_pct: float = 0 + stop_pct: float = 0 + remind_date: str | None = None + lead_days: int = 1 + + +def _reload_engine(request: Request) -> None: + """批次规则保存/删除后重载引擎 — 复用监控规则 API 的共享重载 (含指数纠正)。""" + from app.api.monitor_rules import _sync_engine + + _sync_engine(request) + + +def sync_lot(request: Request, lot: dict) -> None: + """写批次文件 + 同步其两条派生监控规则 + 重载引擎。 + + 派生规则继承用户默认推送渠道 (webhook_default_channels), 否则批次告警会静默只走应用内。 + """ + from app.services import preferences + + data_dir = _data_dir(request) + with _write_lock: + default_channels = preferences.get_webhook_default_channels() + # ETF/指数等资产类型解析 (止盈止损价格规则须走对应资产监控轮才会触发) + asset_type = _resolve_asset_type(request, lot["symbol"]) + price_rule, date_rule = lots_domain.lot_to_rules(lot) + rules_to_write: list[dict] = [] + rules_to_delete: list[str] = [] + for rid, rule in ((f"{lot['id']}_p", price_rule), (f"{lot['id']}_d", date_rule)): + if rule is None: + rules_to_delete.append(rid) + continue + rule["asset_type"] = asset_type + rule.setdefault("webhook_channels", list(default_channels)) + # 保留旧 created_at, 避免编辑批次后派生规则在监控中心列表跳位 + existing = monitor_rules.load_one(data_dir, rid) + if existing and existing.get("created_at"): + rule["created_at"] = existing["created_at"] + try: + monitor_rules.validate(rule) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) from e + rules_to_write.append(monitor_rules.normalize(rule)) + lots_domain.save_one(data_dir, lot) + for rid in rules_to_delete: + monitor_rules.delete_one(data_dir, rid) + for rule in rules_to_write: + monitor_rules.save_one(data_dir, rule) + _reload_engine(request) + + +@router.get("") +def list_lots(request: Request): + return {"lots": lots_domain.load_all(_data_dir(request))} + + +@router.post("") +def upsert_lot(lot_in: LotModel, request: Request): + """新建/更新一个批次。id 缺省时服务端生成 (紧凑, 保证 {id}_p/_d 规则 id ≤ 40 字符)。""" + lot = lot_in.model_dump() + if not lot.get("id"): + lot["id"] = f"lot_{int(time.time() * 1000):x}_{secrets.token_hex(2)}" + try: + lots_domain.validate_lot(lot) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) from e + lot = lots_domain.normalize_lot(lot) + sync_lot(request, lot) + return {"ok": True, "lot": lot} + + +@router.delete("/{lot_id}") +def delete_lot(lot_id: str, request: Request): + if not monitor_rules.ID_RE.match(lot_id): + raise HTTPException(status_code=400, detail="批次 id 非法") + data_dir = _data_dir(request) + with _write_lock: + deleted = lots_domain.delete_one(data_dir, lot_id) + # 两条派生规则都要删 (用 or 会短路跳过第二条) + deleted_p = monitor_rules.delete_one(data_dir, f"{lot_id}_p") + deleted_d = monitor_rules.delete_one(data_dir, f"{lot_id}_d") + if deleted or deleted_p or deleted_d: + _reload_engine(request) + return {"ok": True} diff --git a/backend/app/api/monitor_rules.py b/backend/app/api/monitor_rules.py index 139296e..f92322f 100644 --- a/backend/app/api/monitor_rules.py +++ b/backend/app/api/monitor_rules.py @@ -97,6 +97,9 @@ class RuleModel(BaseModel): conditions: list[ConditionModel] = [] logic: str = "and" # and | or cooldown_seconds: int = 3600 + # date 类型 (日期提醒): 纯日历窗口, 无 conditions + remind_date: str | None = None # YYYY-MM-DD + lead_days: int = 0 # 提前 N 天进入提醒窗口 severity: str = "info" # info | warn | critical webhook_url: str = "" # Webhook 推送地址 (推送到 QMT 等外部软件, 待定) webhook_enabled: bool = False # 兼容老规则 (已由 webhook_channels 取代, 仅做向后兼容读) @@ -166,6 +169,7 @@ def get_options(request: Request): {"key": "abnormal", "label": "异动监控"}, {"key": "sector", "label": "板块监控"}, {"key": "volume_delta", "label": "轮询放量"}, + {"key": "date", "label": "日期提醒"}, ], "scopes": [ {"key": "symbols", "label": "指定标的"}, @@ -288,6 +292,9 @@ def save_rule(req: RuleModel, request: Request): raise HTTPException(status_code=400, detail=str(e)) from e # 编辑现有规则时, 保留原 created_at (避免按时间排序时位置跳动) existing = monitor_rules.load_one(_data_dir(request), rule["id"]) + # 批次派生规则由「持仓提醒」页托管, 监控中心只读 (启停/改/删均回持仓页) + if existing and existing.get("lot_id"): + raise HTTPException(status_code=409, detail="该规则由「持仓提醒」页托管, 请在持仓提醒页修改") if existing and existing.get("created_at"): rule["created_at"] = existing["created_at"] try: @@ -344,6 +351,10 @@ def save_rule(req: RuleModel, request: Request): def delete_rule(rule_id: str, request: Request): if not monitor_rules.ID_RE.match(rule_id): raise HTTPException(status_code=400, detail="规则 id 非法") + # 批次派生规则由「持仓提醒」页托管, 删除需在持仓页操作 (级联清理派生规则) + existing = monitor_rules.load_one(_data_dir(request), rule_id) + if existing and existing.get("lot_id"): + raise HTTPException(status_code=409, detail="该规则由「持仓提醒」页托管, 请在持仓提醒页删除批次") deleted = monitor_rules.delete_one(_data_dir(request), rule_id) if not deleted: raise HTTPException(status_code=404, detail="规则不存在") diff --git a/backend/app/main.py b/backend/app/main.py index 801fa96..647d0a6 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -23,6 +23,7 @@ from app.api import ( indices, intraday, kline, + lots, market_recap, mining, monitor_rules, @@ -471,6 +472,7 @@ app.include_router(settings_api.router) app.include_router(strategy.router) app.include_router(signals.router) app.include_router(monitor_rules.router) +app.include_router(lots.router) app.include_router(alerts.router) app.include_router(rps.router) diff --git a/backend/app/services/fs_utils.py b/backend/app/services/fs_utils.py new file mode 100644 index 0000000..65e2aa7 --- /dev/null +++ b/backend/app/services/fs_utils.py @@ -0,0 +1,16 @@ +"""文件系统小工具 — 原子写等。 + +历史遗留: json_report_store / strategy_cache / kline_sync 等模块里各有一份内联的 +同款原子写。新代码统一用本模块的 atomic_write_text, 一处实现一处维护。 +""" +from __future__ import annotations + +import os +from pathlib import Path + + +def atomic_write_text(path: Path, text: str) -> None: + """临时文件 + os.replace 原子替换, 避免读侧读到半截 JSON。""" + tmp = path.with_suffix(path.suffix + ".tmp") + tmp.write_text(text, encoding="utf-8") + os.replace(tmp, path) diff --git a/backend/app/services/quote_service.py b/backend/app/services/quote_service.py index ce69ac5..da21014 100644 --- a/backend/app/services/quote_service.py +++ b/backend/app/services/quote_service.py @@ -37,6 +37,26 @@ from app.market_time import cn_now, cn_today from app.parquet import scan_daily_parquet from app.services.index_const import CORE_INDEX_SYMBOLS from app.strategy.intraday_signals import IntradaySignalEvaluator +from app.strategy.monitor import format_alert_quote + +# 告警来源 → 中文标签 (webhook 标题 / 系统通知标题共用) +SOURCE_LABELS = { + "strategy": "策略", "signal": "信号", "price": "价格", + "market": "异动", "ladder": "连板梯队", "sector": "板块", + "volume_delta": "放量", "abnormal": "异动", "date": "日期提醒", +} + + +def _body_with_quote(body: str, ev: dict) -> str: + """推送正文尾部补上触发时的现价/涨跌幅 (日期提醒无行情, 自然为空)。 + + 默认告警的 message 已由引擎拼过引语 (monitor._default_message), 这里仅在正文 + 尚未带引语时追加, 避免「现价」出现两遍 (自定义 message 的规则则补上这一句)。 + """ + quote_tail = format_alert_quote(ev.get("price"), ev.get("change_pct")) + if not quote_tail or body.endswith(quote_tail): + return body + return f"{body} · {quote_tail}" logger = logging.getLogger(__name__) @@ -1063,6 +1083,12 @@ class QuoteService: rule_events += engine.evaluate_abnormal(_overview.get("rows") or []) except Exception as e: # noqa: BLE001 logger.warning("异动监控规则评估失败 (不影响其他告警): %s", e) + # 日期提醒轮: 纯日历、无行情, 已在盘中; 引擎内按天 cooldown 保证每天一次 + if engine.has_rule_type("date"): + try: + rule_events = rule_events + engine.evaluate_date_rules() + except Exception as e: # noqa: BLE001 + logger.warning("日期提醒评估失败 (不影响其他告警): %s", e) # ETF 规则轮: 股票快照不含 ETF, 用 ETF enriched 快照单独评估。 # 独立 try —— ETF 轮任何异常都不得丢弃本轮已算出的股票告警。 # refresh=False —— 不在轮询线程上触发 ETF 冷缓存的同步重算 (缓存由 ETF 实时 @@ -1427,11 +1453,6 @@ class QuoteService: return # 反查规则, 过滤出启用推送的事件 - source_labels = { - "strategy": "策略", "signal": "信号", - "price": "价格", "market": "异动", "ladder": "连板梯队", - "sector": "板块", "volume_delta": "放量", - } rules = engine.rules if engine is not None else {} enqueued = 0 for ev in rule_events: @@ -1442,12 +1463,14 @@ class QuoteService: if not channels: continue source = ev.get("source", "") - source_label = source_labels.get(source, source or "通知") + source_label = SOURCE_LABELS.get(source, source or "通知") symbol = ev.get("symbol") or "" name = ev.get("name") or "" message = ev.get("message") or "" title = source_label body = f"{symbol} {name} {message}".strip() if symbol else (message or name) + # 补上触发时的现价/涨跌幅, 让推送可执行 (止损到底触发在哪个价位) + body = _body_with_quote(body, ev) # 提交到独立线程池, 不阻塞行情轮询线程 (webhook 慢/重试不拖累实时行情+告警)。 # 按渠道独立投递: 飞书 / 企业微信谁被勾选且已配置就推谁。 # 应用内 alerts.jsonl 记录与 SSE 已在前面完成, 不依赖 webhook 成败, @@ -1481,10 +1504,7 @@ class QuoteService: for ev in all_alerts: # 通知标题: 用 source 分类 (策略/信号/价格/异动) source = ev.get("source", "") - source_label = { - "strategy": "策略", "signal": "信号", - "price": "价格", "market": "异动", "sector": "板块", - }.get(source, source or "通知") + source_label = SOURCE_LABELS.get(source, source or "通知") name = ev.get("name") or "" symbol = ev.get("symbol") or "" @@ -1495,6 +1515,8 @@ class QuoteService: body = f"{symbol} {name} {message}".strip() else: body = message or name + # 补上触发时的现价/涨跌幅 (日期提醒无行情, 自然为空) + body = _body_with_quote(body, ev) title = f"TickFlow · {source_label}" notify_adapter.notify(title, body) diff --git a/backend/app/strategy/lots.py b/backend/app/strategy/lots.py new file mode 100644 index 0000000..393b7dc --- /dev/null +++ b/backend/app/strategy/lots.py @@ -0,0 +1,177 @@ +"""批次登记域 — 薄"批次"页 (持仓提醒): 只生成监控规则, 不做会计。 + +每行一个买入批次 → 派生两条规则: lot_{id}_p (price 止盈止损) / lot_{id}_d (date 到期提醒)。 +记账/加减仓属"交易口径", 不在本模块 (issue #230)。纯函数 + 文件存储, 镜像 monitor_rules.py, +不做 API、不做引擎重载。 +""" +from __future__ import annotations + +import json +import logging +from datetime import UTC, datetime +from datetime import date as _date +from pathlib import Path + +from app.services.fs_utils import atomic_write_text +from app.strategy import monitor_rules +from app.strategy.monitor import MonitorRuleEngine # 复用条件文本拼装 (静态方法) + +logger = logging.getLogger(__name__) + +# id 需满足规则 id 同款正则, 且为后缀留位: 派生规则 {id}_p/_d 不得超过 40 字符 +_ID = monitor_rules.ID_RE +_MAX_ID_LEN = 40 - 2 # 派生规则 id 后缀 "_p" / "_d" + + +def _dir(data_dir: Path) -> Path: + d = data_dir / "user_data" / "lots" + d.mkdir(parents=True, exist_ok=True) + return d + + +def _path(data_dir: Path, lot_id: str) -> Path: + return _dir(data_dir) / f"{lot_id}.json" + + +def _now_iso() -> str: + return datetime.now(UTC).isoformat() + + +# ── 校验与归一化 ──────────────────────────────────────── +def validate_lot(lot: dict) -> None: + """校验批次字段, 非法抛 ValueError (中文信息)。""" + lot_id = lot.get("id") + if lot_id is not None and ( + not isinstance(lot_id, str) or not _ID.match(lot_id) or len(lot_id) > _MAX_ID_LEN + ): + raise ValueError(f"批次 id 非法 (仅小写字母数字下划线, 且需为派生规则 id 留位): {lot_id!r}") + if not (lot.get("symbol") or "").strip(): + raise ValueError("symbol 不能为空") + cost = lot.get("cost_price") + if isinstance(cost, bool) or not isinstance(cost, (int, float)) or cost <= 0: + raise ValueError("cost_price 必须是正数") + for key, label in (("qty", "数量"), ("target_pct", "止盈%"), ("stop_pct", "止损%")): + v = lot.get(key, 0) + if isinstance(v, bool) or not isinstance(v, (int, float)) or v < 0: + raise ValueError(f"{label} 不能为负数") + lead = lot.get("lead_days", 0) + if isinstance(lead, bool) or not isinstance(lead, int) or lead < 0: + raise ValueError("lead_days 必须是非负整数") + for key, label in (("buy_date", "买入日期"), ("remind_date", "到期日")): + raw = lot.get(key) + if raw not in (None, ""): + try: + _date.fromisoformat(raw) + except ValueError: + raise ValueError(f"{label} 必须是 YYYY-MM-DD: {raw!r}") from None + if not (lot.get("target_pct", 0) > 0 or lot.get("stop_pct", 0) > 0 or lot.get("remind_date")): + raise ValueError("止盈% / 止损% / 到期日 至少设置一项 (否则无监控点)") + + +def normalize_lot(lot: dict) -> dict: + """补全默认字段, 返回规范化后的批次 (不校验)。""" + d = dict(lot) + d["symbol"] = (d.get("symbol") or "").strip() + d.setdefault("qty", 0) + d.setdefault("cost_price", 0) + d.setdefault("buy_date", None) + d.setdefault("target_pct", 0) + d.setdefault("stop_pct", 0) + d.setdefault("remind_date", None) + d.setdefault("lead_days", 1) + d.setdefault("created_at", _now_iso()) + return d + + +# ── 持久化 ───────────────────────────────────────────── +def load_all(data_dir: Path) -> list[dict]: + """读取全部批次。损坏的文件被跳过。""" + out: list[dict] = [] + for f in sorted(_dir(data_dir).glob("lot_*.json")): + try: + out.append(normalize_lot(json.loads(f.read_text(encoding="utf-8")))) + except Exception as e: + logger.warning("lot load failed %s: %s", f.name, e) + return out + + +def save_one(data_dir: Path, lot: dict) -> None: + p = _path(data_dir, lot["id"]) + p.parent.mkdir(parents=True, exist_ok=True) + atomic_write_text(p, json.dumps(lot, ensure_ascii=False, indent=2)) + + +def delete_one(data_dir: Path, lot_id: str) -> bool: + p = _path(data_dir, lot_id) + if p.exists(): + p.unlink() + return True + return False + + +# ── 批次 → 监控规则 (纯映射) ─────────────────────────── +def lot_to_rules(lot: dict) -> tuple[dict | None, dict | None]: + """批次 → (price 止盈止损规则, date 到期规则); 无对应监控点时返回 None。 + + 纯映射不做 I/O; 规则 id 派生自批次 id ({lot_id}_p/_d), 保证稳定可级联。 + """ + symbol = lot["symbol"] + lot_id = lot["id"] + cost = float(lot["cost_price"]) + target = float(lot.get("target_pct", 0)) + stop = float(lot.get("stop_pct", 0)) + qty = float(lot.get("qty", 0) or 0) + qty_text = f" · {qty:g}股" if qty > 0 else "" + + conds: list[dict] = [] + if target > 0: + conds.append({"field": "close", "op": ">=", "value": round(cost * (1 + target / 100), 4)}) + if stop > 0: + conds.append({"field": "close", "op": "<=", "value": round(cost * (1 - stop / 100), 4)}) + price_rule = None + if conds: + msg = f"批次止盈止损 · 成本{cost:g}" + if target > 0: + msg += f" · 止盈{target:g}%" + if stop > 0: + msg += f" · 止损{stop:g}%" + msg += qty_text + cond_text = MonitorRuleEngine._format_conditions_text({"logic": "or"}, conds) + if cond_text: + msg += f" · {cond_text}" + price_rule = { + "id": f"{lot_id}_p", + "name": f"批次止盈止损 · {symbol}", + "type": "price", + "asset_type": "stock", + "scope": "symbols", + "symbols": [symbol], + "conditions": conds, + "logic": "or", + "cooldown_seconds": 86400, + "severity": "warn", + "message": msg, + "enabled": True, + "lot_id": lot_id, + } + + date_rule = None + if lot.get("remind_date"): + lead = int(lot.get("lead_days", 1)) + date_rule = { + "id": f"{lot_id}_d", + "name": f"批次到期 · {symbol}", + "type": "date", + "asset_type": "stock", + "scope": "symbols", + "symbols": [symbol], + "remind_date": lot["remind_date"], + "lead_days": lead, + "cooldown_seconds": 86400, + "severity": "info", + # 提前天数由引擎 evaluate_date_rules 统一追加, 这里只放静态部分 + "message": f"批次到期提醒 · {lot['remind_date']}{qty_text}", + "enabled": True, + "lot_id": lot_id, + } + return price_rule, date_rule diff --git a/backend/app/strategy/monitor.py b/backend/app/strategy/monitor.py index a6afdfd..2e736cd 100644 --- a/backend/app/strategy/monitor.py +++ b/backend/app/strategy/monitor.py @@ -25,6 +25,7 @@ from app.market_time import cn_today from app.strategy import config as _strategy_config from app.strategy.custom_signals import _OP_BUILDERS # type: ignore # 复用运算符构造器 from app.strategy.intraday_signals import INTRADAY_SIGNAL_LABELS, uses_intraday_signals +from app.strategy.monitor_rules import date_rule_in_window logger = logging.getLogger(__name__) @@ -71,6 +72,17 @@ def _signal_cn_name(name: str) -> str: return _SIGNAL_CN.get(name, name) +def format_alert_quote(price, change_pct) -> str: + """告警正文尾部: '现价 1650.0 · +10.0%'。price/pct 均可缺; pct 为小数制。""" + parts = [] + if price is not None: + parts.append(f"现价 {price}") + if change_pct is not None: + sign = "+" if change_pct >= 0 else "" + parts.append(f"{sign}{change_pct * 100:.1f}%") + return " · ".join(parts) + + @dataclass class StrategyAlert: """策略告警""" @@ -323,6 +335,10 @@ class MonitorRuleEngine: self._rules: dict[str, dict] = {} # rule_id → rule # (rule_id, symbol, event_type) → 上次触发时间戳(秒)。用于 cooldown 去重。 self._last_fire: dict[tuple[str, str, str], float] = {} + # date 规则每个交易日只在首个轮询评估一次; 规则集变更时失效重评 + self._date_eval_day: str | None = None + self._date_eval_rules_version = -1 + self._rules_version = 0 # set/add/remove/clear 递增, 供 date 缓存失效 self._strategy_engine = None # 延迟注入, type=strategy 规则用它跑选股 # symbol → 股票名 (enriched DataFrame 已 drop name 列, 触发时从此映射回填) self._name_map: dict[str, str] = {} @@ -428,6 +444,8 @@ class MonitorRuleEngine: rule.get("threshold_pct"), rule.get("window_minutes"), rule.get("abnormal_window"), + rule.get("remind_date"), + rule.get("lead_days"), ) def set_rules(self, rules: list[dict]) -> None: @@ -476,12 +494,14 @@ class MonitorRuleEngine: if key[0] in active_ids } logger.info("MonitorRuleEngine: 装载 %d 条规则", len(self._rules)) + self._rules_version += 1 def add_rule(self, rule: dict) -> None: if rule.get("enabled") is not False: self._rules[rule["id"]] = rule else: self._rules.pop(rule["id"], None) + self._rules_version += 1 def remove_rule(self, rule_id: str) -> None: self._rules.pop(rule_id, None) @@ -498,6 +518,7 @@ class MonitorRuleEngine: self._sector_condition_state = { k: v for k, v in self._sector_condition_state.items() if k[0] != rule_id } + self._rules_version += 1 def clear(self) -> None: self._rules.clear() @@ -506,6 +527,7 @@ class MonitorRuleEngine: self._strategy_signal_state.clear() self._strategy_signal_seen.clear() self._sector_condition_state.clear() + self._rules_version += 1 @property def rules(self) -> dict[str, dict]: @@ -669,7 +691,9 @@ class MonitorRuleEngine: for rule_id, rule in list(self._rules.items()): if rule.get("asset_type", "stock") != asset_type: continue - if rule.get("type") in ("sector", "abnormal"): + if rule.get("type") in ("sector", "abnormal", "date"): + # 三者不走行情 DataFrame 评估, 各走 evaluate_sectors / evaluate_abnormal / + # evaluate_date_rules 专用路径 continue try: events.extend(self._evaluate_rule(df, rule, now)) @@ -683,6 +707,80 @@ class MonitorRuleEngine: return events + def evaluate_date_rules(self, now: float | None = None) -> list[dict]: + """纯日历评估 date 规则: 窗口命中 + 每天最多一次, 无行情条件。 + + 由行情轮询在盘中调用 (quote_service._evaluate_monitors), 事件与 _evaluate_rule 同构。 + 窗口按自然日; 到期落在休市/节假日时需 lead_days 覆盖 (交易日历口径待 issue 定夺)。 + 每个交易日只在首个轮询完整评估一次, 其余轮次命中缓存直接跳过。 + """ + now = now if now is not None else time.time() + today_iso = cn_today().isoformat() + if self._date_eval_day == today_iso and self._date_eval_rules_version == self._rules_version: + return [] + # 跨天首轮清掉已过期日期的按天 cooldown 键, 避免 _last_fire 无限累积 + self._last_fire = { + key: value + for key, value in self._last_fire.items() + if not (key[1].startswith("_date_") and key[1] != f"_date_{today_iso}") + } + + today_d = _dt.date.fromisoformat(today_iso) + events: list[dict] = [] + for rule in list(self._rules.values()): + if rule.get("type") != "date" or rule.get("enabled") is False: + continue + remind = rule.get("remind_date") or "" + if not date_rule_in_window(remind, int(rule.get("lead_days", 0)), today_iso): + continue + # 按天隔离: 窗口内每天最多触发一次 + key = (rule["id"], f"_date_{today_iso}", "date") + cooldown = int(rule.get("cooldown_seconds") or 86400) + last = self._last_fire.get(key) + if last is not None and (now - last) < cooldown: + continue + self._last_fire[key] = now + + symbols = [s for s in rule.get("symbols", []) if s] + single_symbol = symbols[0] if len(symbols) == 1 else None + msg = rule.get("message") or f"日期提醒 · {today_iso}" + try: + remain = (_dt.date.fromisoformat(remind) - today_d).days + except ValueError: + remain = 0 + msg += " · 今日到期" if remain <= 0 else f" · {remain}天后到期" + # 单标的由 ev.symbol 携带; 仅多标的时拼列表 + if len(symbols) > 1: + shown = "、".join(symbols[:3]) + ("等" if len(symbols) > 3 else "") + msg = f"{msg} · {shown}" + + ev = { + "ts": int(now * 1000), + "rule_id": rule["id"], + "rule_name": rule.get("name", ""), + "strategy_id": None, + "source": "date", + "type": "date_reminder", + "symbol": single_symbol or "", + "name": (self._name_map.get(single_symbol) or single_symbol) if single_symbol else None, + "message": msg, + "price": None, + "change_pct": None, + "signals": [], + "severity": rule.get("severity", "info"), + "conditions": [], + "logic": "and", + } + events.append(ev) + if self._alert_handler: + try: + self._alert_handler(ev) + except Exception as e: # noqa: BLE001 + logger.warning("alert handler failed: %s", e) + self._date_eval_day = today_iso + self._date_eval_rules_version = self._rules_version + return events + def evaluate_sectors( self, stock_df: pl.DataFrame, @@ -1623,12 +1721,7 @@ class MonitorRuleEngine: # signal / price / market: 条件摘要 + 现价 + 涨跌幅 # 条件摘要: 把 conditions (truth/比较) 拼成可读串, 如 "MA20金叉 且 量比>2" cond_text = self._format_conditions_text(rule, conditions) - price_text = f"现价 {price}" if price is not None else "" - pct_text = "" - if pct is not None: - sign = "+" if pct >= 0 else "" - pct_text = f"{sign}{pct * 100:.1f}%" - tail = " · ".join(s for s in (price_text, pct_text) if s) + tail = format_alert_quote(price, pct) if cond_text and tail: return f"{cond_text} · {tail}" return cond_text or tail or "监控触发" diff --git a/backend/app/strategy/monitor_rules.py b/backend/app/strategy/monitor_rules.py index 2d6571b..972e276 100644 --- a/backend/app/strategy/monitor_rules.py +++ b/backend/app/strategy/monitor_rules.py @@ -18,9 +18,10 @@ import json import logging import math import re -from datetime import datetime, timezone +from datetime import date, datetime, timedelta, timezone from pathlib import Path +from app.services.fs_utils import atomic_write_text from app.strategy.custom_signals import ALLOWED_FIELDS from app.strategy.intraday_signals import uses_intraday_signals @@ -28,7 +29,7 @@ logger = logging.getLogger(__name__) # ── 常量 ──────────────────────────────────────────────── ID_RE = re.compile(r"^[a-z0-9_]{1,40}$") -RULE_TYPES = {"strategy", "signal", "price", "market", "ladder", "sector", "abnormal", "volume_delta"} +RULE_TYPES = {"strategy", "signal", "price", "market", "ladder", "sector", "abnormal", "volume_delta", "date"} SCOPES = {"symbols", "all", "sector", "watchlist_group"} LOGICS = {"and", "or"} DIRECTIONS = {"entry", "exit", "both"} @@ -100,7 +101,7 @@ def load_one(data_dir: Path, rule_id: str) -> dict | None: def save_one(data_dir: Path, rule: dict) -> None: p = _path(data_dir, rule["id"]) p.parent.mkdir(parents=True, exist_ok=True) - p.write_text(json.dumps(rule, ensure_ascii=False, indent=2), encoding="utf-8") + atomic_write_text(p, json.dumps(rule, ensure_ascii=False, indent=2)) def delete_one(data_dir: Path, rule_id: str) -> bool: @@ -117,6 +118,22 @@ def _is_signal_field(field: str) -> bool: return any(field.startswith(p) for p in _SIGNAL_PREFIXES) +def date_rule_in_window(remind_date: str, lead_days: int, today: str) -> bool: + """提醒窗口 [remind_date - lead_days, remind_date] 是否包含 today (均 YYYY-MM-DD)。 + + 只判自然日历窗口; 是否在交易时段由调用方决定。到期落在休市/节假日不会顺延, + 需 lead_days 覆盖 (交易日历口径待 issue 定夺)。非法输入一律返回 False (fail-safe)。 + """ + try: + remind = date.fromisoformat(remind_date) + today_d = date.fromisoformat(today) + lead = max(0, int(lead_days or 0)) + except (ValueError, TypeError): + return False + start = remind - timedelta(days=lead) + return start <= today_d <= remind + + def validate(rule: dict) -> None: """校验一条监控规则,非法则抛 ValueError (含中文信息)。""" rid = rule.get("id", "") @@ -236,6 +253,20 @@ def validate(rule: dict) -> None: raise ValueError(f"basic_filter.{key} 必须是正数字或 null") else: raise ValueError(f"basic_filter 不支持字段: {key}") + elif rule.get("type") == "date": + # 日期提醒: 纯日历, 锚定标的 (scope=symbols) 避免无对象的空提醒 + remind = rule.get("remind_date") + if not isinstance(remind, str) or not remind.strip(): + raise ValueError("日期提醒规则必须指定 remind_date") + try: + date.fromisoformat(remind.strip()) + except ValueError: + raise ValueError(f"remind_date 必须是 YYYY-MM-DD 日期: {remind!r}") + lead = rule.get("lead_days", 0) + if isinstance(lead, bool) or not isinstance(lead, int) or lead < 0: + raise ValueError("lead_days 必须是非负整数 (提前提醒天数)") + if rule.get("conditions"): + raise ValueError("日期提醒规则不支持行情 conditions") else: # 信号/价格/市场类型: 需要 conditions conds = rule.get("conditions") @@ -349,6 +380,12 @@ def normalize(rule: dict) -> dict: r["scope"] = "all" r["symbols"] = [] r["group_id"] = None + # date 专属默认字段 (日期提醒): 纯日历窗口, 无行情条件, 每天至多一次 + if r.get("type") == "date": + r["conditions"] = [] + r.setdefault("remind_date", None) + r["lead_days"] = int(r.get("lead_days") or 0) + r["cooldown_seconds"] = 86400 # abnormal 专属默认字段 (异动边缘监控) r.setdefault("abnormal_window", "any") r.setdefault("logic", "and") diff --git a/backend/tests/test_date_rule.py b/backend/tests/test_date_rule.py new file mode 100644 index 0000000..7a1c00e --- /dev/null +++ b/backend/tests/test_date_rule.py @@ -0,0 +1,151 @@ +"""日期提醒 (date) 规则测试: 窗口判定 + 引擎按天去重 + 校验/归一化。 + +date 规则是纯日历、无行情条件, 由 quote_service 盘中轮询调用 +MonitorRuleEngine.evaluate_date_rules()。这里只测引擎与规则存储层。 +""" +from __future__ import annotations + +import polars as pl +import pytest + +from app.strategy import monitor_rules +from app.strategy.monitor import MonitorRuleEngine, format_alert_quote + + +def _rule(**kw): + r = { + "id": "date1", "name": "批次到期", "type": "date", + "asset_type": "stock", "scope": "symbols", + "symbols": ["600519.SH"], + "remind_date": "2026-08-30", "lead_days": 3, + "cooldown_seconds": 86400, "severity": "info", "enabled": True, + "message": "批次到期提醒 · 2026-08-30", + } + r.update(kw) + return r + + +# ── 窗口纯函数 ────────────────────────────────────────── +def test_window_pure_unit_cases(): + w = monitor_rules.date_rule_in_window + assert w("2026-08-30", 3, "2026-08-28") is True # 窗口左界内 + assert w("2026-08-30", 3, "2026-08-27") is True # 恰在左界 (含) + assert w("2026-08-30", 3, "2026-08-26") is False # 早于左界 + assert w("2026-08-30", 3, "2026-08-30") is True # 到期当天 + assert w("2026-08-30", 3, "2026-08-31") is False # 过期 + assert w("2026-08-30", 0, "2026-08-30") is True # 不提前 + assert w("2026-08-30", 0, "2026-08-29") is False + assert w("not-a-date", 3, "2026-08-28") is False # 非法输入 fail-safe + assert w("2026-08-30", "x", "2026-08-28") is False + + +def test_validate_and_normalize_date(): + valid = {"id": "date2", "name": "n", "type": "date", "scope": "symbols", + "symbols": ["600519.SH"], "remind_date": "2026-09-01", "lead_days": 2} + monitor_rules.validate(valid) + n = monitor_rules.normalize(valid) + assert n["cooldown_seconds"] == 86400 # 每天最多一次 + assert n["conditions"] == [] + assert n["scope"] == "symbols" + + # 归一化后为空 symbols 仍合法 (手动建), 但校验拒绝空 symbols (锚定标的) + with pytest.raises(ValueError): + monitor_rules.validate({**valid, "symbols": []}) + + with pytest.raises(ValueError): + monitor_rules.validate({**valid, "remind_date": None}) # 必填 + with pytest.raises(ValueError): + monitor_rules.validate({**valid, "remind_date": "2026/09/01"}) # 格式 + with pytest.raises(ValueError): + monitor_rules.validate({**valid, "lead_days": -1}) # 非负 + with pytest.raises(ValueError): + monitor_rules.validate({**valid, "conditions": [{"field": "close", "op": ">=", "value": 10.0}]}) + assert "date" in monitor_rules.RULE_TYPES + + +# ── 引擎评估 ──────────────────────────────────────────── +def _monkey_today(monkeypatch, iso: str): + from datetime import date + + monkeypatch.setattr("app.strategy.monitor.cn_today", lambda: date.fromisoformat(iso)) + + +def test_date_engine_fires_in_window(monkeypatch): + _monkey_today(monkeypatch, "2026-08-28") + eng = MonitorRuleEngine() + eng.set_rules([_rule()]) + evs = eng.evaluate_date_rules() + assert len(evs) == 1 + ev = evs[0] + assert ev["source"] == "date" + assert ev["type"] == "date_reminder" + assert ev["symbol"] == "600519.SH" + assert "2天后到期" in ev["message"] # 2026-08-30 - 今天(08-28) + assert ev["message"] # 单标的: 不再把 symbol 拼进 message + assert "600519.SH" not in ev["message"] + assert ev["price"] is None and ev["change_pct"] is None # 日期提醒无行情 + assert ev["conditions"] == [] and ev["logic"] == "and" + + +def test_date_engine_today_expiry_wording(monkeypatch): + _monkey_today(monkeypatch, "2026-08-30") + eng = MonitorRuleEngine() + eng.set_rules([_rule(lead_days=0)]) + evs = eng.evaluate_date_rules() + assert len(evs) == 1 + assert "今日到期" in evs[0]["message"] + + +def test_date_engine_skips_outside_window(monkeypatch): + _monkey_today(monkeypatch, "2026-08-31") # 过期 + eng = MonitorRuleEngine() + eng.set_rules([_rule()]) + assert eng.evaluate_date_rules() == [] + + +def test_date_engine_ignores_disabled_and_non_date(monkeypatch): + _monkey_today(monkeypatch, "2026-08-28") + eng = MonitorRuleEngine() + eng.set_rules([_rule(enabled=False), _rule(id="price1", type="price", conditions=[ + {"field": "close", "op": ">=", "value": 100.0}, + ], message="")]) + assert eng.evaluate_date_rules() == [] + + +def test_date_engine_once_per_day(monkeypatch): + _monkey_today(monkeypatch, "2026-08-28") + eng = MonitorRuleEngine() + eng.set_rules([_rule()]) + assert len(eng.evaluate_date_rules()) == 1 + assert eng.evaluate_date_rules() == [] # 同日 cooldown 去重 + # 跨天: 窗口内仍命中 → 再提醒一次 + _monkey_today(monkeypatch, "2026-08-29") + assert len(eng.evaluate_date_rules()) == 1 + + +def test_date_engine_alerts_handler_called(monkeypatch): + _monkey_today(monkeypatch, "2026-08-28") + seen: list[dict] = [] + eng = MonitorRuleEngine(alert_handler=seen.append) + eng.set_rules([_rule()]) + eng.evaluate_date_rules() + assert len(seen) == 1 and seen[0]["source"] == "date" + + +def test_date_not_evaluated_by_quote_evaluate(monkeypatch): + """date 规则不走 evaluate(df) 主循环, 避免与 date_rule 专用路径双触发。""" + _monkey_today(monkeypatch, "2026-08-28") + eng = MonitorRuleEngine() + eng.set_rules([_rule()]) + df = pl.DataFrame({"symbol": ["600519.SH"], "close": [1500.0]}) + assert eng.evaluate(df) == [] + assert len(eng.evaluate_date_rules()) == 1 + + +# ── 告警引语 ──────────────────────────────────────────── +def test_format_alert_quote(): + assert format_alert_quote(1500.0, 0.1) == "现价 1500.0 · +10.0%" + assert format_alert_quote(1425.0, -0.05) == "现价 1425.0 · -5.0%" + assert format_alert_quote(None, None) == "" # 日期提醒等无行情 + assert format_alert_quote(None, 0.03) == "+3.0%" + assert format_alert_quote(10.0, None) == "现价 10.0" diff --git a/backend/tests/test_lots.py b/backend/tests/test_lots.py new file mode 100644 index 0000000..56d8801 --- /dev/null +++ b/backend/tests/test_lots.py @@ -0,0 +1,198 @@ +"""批次 (持仓提醒) 测试: 批次→规则映射 + 校验 + sync 一致性 (锁/校验先行/级联删除)。""" +from __future__ import annotations + +from pathlib import Path +from types import SimpleNamespace + +import pytest +from fastapi import HTTPException + +from app.api import lots as lots_api +from app.strategy import lots as lots_domain +from app.strategy import monitor_rules + + +def _lot(**overrides): + lot = { + "id": "lot_test1", + "symbol": "600519.SH", + "qty": 100, + "cost_price": 1500.0, + "buy_date": "2026-08-01", + "target_pct": 10, + "stop_pct": 5, + "remind_date": "2026-09-01", + "lead_days": 2, + } + lot.update(overrides) + return lot + + +class _EngineStub: + def __init__(self) -> None: + self.set_calls = 0 + self.rules = [] + + def set_rules(self, rules) -> None: + self.set_calls += 1 + self.rules = rules + + +def _make_request(tmp_path: Path, engine: _EngineStub): + repo = SimpleNamespace( + store=SimpleNamespace(data_dir=tmp_path), + resolve_asset_type=lambda _symbol: "stock", + ) + state = SimpleNamespace(repo=repo, monitor_engine=engine) + return SimpleNamespace(app=SimpleNamespace(state=state)) + + +def _lot_path(tmp_path: Path, lot_id: str) -> Path: + return tmp_path / "user_data" / "lots" / f"{lot_id}.json" + + +def _patch_channels(monkeypatch) -> None: + monkeypatch.setattr("app.services.preferences.get_webhook_default_channels", lambda: []) + + +# ── 纯映射: 批次 → 规则 ──────────────────────────────── +def test_lot_to_rules_price_and_date(): + price, date_rule = lots_domain.lot_to_rules(_lot()) + assert price is not None and date_rule is not None + assert price["id"] == "lot_test1_p" and price["type"] == "price" + assert price["scope"] == "symbols" and price["symbols"] == ["600519.SH"] + assert price["conditions"] == [ + {"field": "close", "op": ">=", "value": 1650.0}, # 1500*1.10 + {"field": "close", "op": "<=", "value": 1425.0}, # 1500*0.95 + ] + assert price["logic"] == "or" + assert price["cooldown_seconds"] == 86400 + assert price["lot_id"] == "lot_test1" + assert "止盈10%" in price["message"] and "止损5%" in price["message"] + assert date_rule["id"] == "lot_test1_d" and date_rule["type"] == "date" + assert date_rule["remind_date"] == "2026-09-01" and date_rule["lead_days"] == 2 + + +def test_lot_to_rules_optional_parts(): + # 只有止盈 (无止损/无到期) → 无 date 规则 + price, date_rule = lots_domain.lot_to_rules(_lot(stop_pct=0, remind_date=None)) + assert price is not None and date_rule is None + assert len(price["conditions"]) == 1 + # 只有到期 (无止盈/止损) → 无 price 规则 + price, date_rule = lots_domain.lot_to_rules(_lot(target_pct=0, stop_pct=0)) + assert price is None and date_rule is not None + + +def test_validate_lot_rules_and_errors(): + lots_domain.validate_lot(lots_domain.normalize_lot(_lot())) + for overrides in ( + {"symbol": " "}, + {"cost_price": 0}, + {"qty": -1}, + {"target_pct": -1}, + {"stop_pct": -1}, + {"lead_days": -1}, + {"remind_date": "2026/09/01"}, + {"buy_date": "bad"}, + {"target_pct": 0, "stop_pct": 0, "remind_date": None}, + ): + with pytest.raises(ValueError): + lots_domain.validate_lot({**_lot(), **overrides}) + + +def test_normalize_lot_defaults(): + n = lots_domain.normalize_lot({"id": "lot_x", "symbol": "600519.SH", "cost_price": 10}) + assert n["qty"] == 0 and n["lead_days"] == 1 + assert n["created_at"] + assert n["symbol"] == "600519.SH" + + +# ── sync_lot 一致性 (校验先行 / 级联 / 单次重载) ───────── +def test_sync_lot_validates_rules_before_write(tmp_path, monkeypatch): + engine = _EngineStub() + request = _make_request(tmp_path, engine) + _patch_channels(monkeypatch) + reloaded = [] + monkeypatch.setattr(lots_api, "_reload_engine", lambda r: reloaded.append(1)) + + def boom(_rule) -> None: + raise ValueError("bad rule") + + monkeypatch.setattr("app.strategy.monitor_rules.validate", boom) + with pytest.raises(HTTPException) as ei: + lots_api.sync_lot(request, _lot()) + assert ei.value.status_code == 400 + # 批次文件、规则文件、引擎重载 都不该发生 (避免半成品) + assert not _lot_path(tmp_path, "lot_test1").exists() + assert reloaded == [] + rules_dir = tmp_path / "user_data" / "monitor_rules" + assert not rules_dir.exists() or list(rules_dir.glob("*.json")) == [] + + +def test_sync_lot_writes_lot_rules_and_reloads_once(tmp_path, monkeypatch): + engine = _EngineStub() + request = _make_request(tmp_path, engine) + _patch_channels(monkeypatch) + reloaded = [] + monkeypatch.setattr(lots_api, "_reload_engine", lambda r: reloaded.append(1)) + + lots_api.sync_lot(request, _lot()) + assert _lot_path(tmp_path, "lot_test1").exists() + price = monitor_rules.load_one(tmp_path, "lot_test1_p") + date_rule = monitor_rules.load_one(tmp_path, "lot_test1_d") + assert price is not None and price["lot_id"] == "lot_test1" + assert price["conditions"][0] == {"field": "close", "op": ">=", "value": 1650.0} + assert date_rule is not None and date_rule["remind_date"] == "2026-09-01" + assert len(reloaded) == 1 + + +def test_sync_lot_removes_rules_when_monitor_point_removed(tmp_path, monkeypatch): + engine = _EngineStub() + request = _make_request(tmp_path, engine) + _patch_channels(monkeypatch) + monkeypatch.setattr(lots_api, "_reload_engine", lambda r: None) + lots_api.sync_lot(request, _lot()) + # 编辑后只剩止盈, 无到期 → date 规则应被级联删除 + lots_api.sync_lot(request, _lot(remind_date=None)) + assert monitor_rules.load_one(tmp_path, "lot_test1_d") is None + assert monitor_rules.load_one(tmp_path, "lot_test1_p") is not None + + +def test_delete_lot_removes_lot_and_both_rules(tmp_path, monkeypatch): + engine = _EngineStub() + request = _make_request(tmp_path, engine) + _patch_channels(monkeypatch) + monkeypatch.setattr(lots_api, "_reload_engine", lambda r: None) + lots_api.sync_lot(request, _lot()) + lots_api.delete_lot("lot_test1", request) + assert not _lot_path(tmp_path, "lot_test1").exists() + assert monitor_rules.load_one(tmp_path, "lot_test1_p") is None + assert monitor_rules.load_one(tmp_path, "lot_test1_d") is None + + +def test_sync_lot_etf_resolves_asset_type(tmp_path, monkeypatch): + engine = _EngineStub() + repo = SimpleNamespace( + store=SimpleNamespace(data_dir=tmp_path), + resolve_asset_type=lambda _symbol: "etf", + ) + state = SimpleNamespace(repo=repo, monitor_engine=engine) + request = SimpleNamespace(app=SimpleNamespace(state=state)) + _patch_channels(monkeypatch) + monkeypatch.setattr(lots_api, "_reload_engine", lambda r: None) + lots_api.sync_lot(request, _lot(symbol="510300.SH")) + # 止盈止损价格规则须走 ETF 监控轮才会触发, 故 asset_type 必须为 etf + price = monitor_rules.load_one(tmp_path, "lot_test1_p") + date_rule = monitor_rules.load_one(tmp_path, "lot_test1_d") + assert price is not None and price["asset_type"] == "etf" + assert date_rule is not None and date_rule["asset_type"] == "etf" + + +def test_upsert_lot_invalid_returns_400(tmp_path, monkeypatch): + engine = _EngineStub() + request = _make_request(tmp_path, engine) + _patch_channels(monkeypatch) + with pytest.raises(HTTPException) as ei: + lots_api.upsert_lot(lots_api.LotModel(**_lot(cost_price=0)), request) + assert ei.value.status_code == 400 + assert "cost_price" in str(ei.value.detail) diff --git a/frontend/src/components/DateShortcuts.tsx b/frontend/src/components/DateShortcuts.tsx new file mode 100644 index 0000000..88b733a --- /dev/null +++ b/frontend/src/components/DateShortcuts.tsx @@ -0,0 +1,53 @@ +import { cn } from '@/lib/cn' +import { fmtDate } from '@/lib/format' + +export interface DateShortcutOption { + label: string + /** 相对基准日期 base 的天数偏移 (0=base, 缺省今天) */ + days: number +} + +interface Props { + /** 当前值 (YYYY-MM-DD), 用于高亮命中的快捷项 */ + value: string + onChange: (v: string) => void + options: DateShortcutOption[] + /** 基准日期 (YYYY-MM-DD), 缺省今天; 如「到期日 = 买入日期 + N 天」 */ + base?: string +} + +function addDaysISO(days: number, base?: string): string { + const d = base ? new Date(`${base}T00:00:00`) : new Date() + d.setDate(d.getDate() + days) + return fmtDate(d) +} + +/** + * 快捷日期 chips — 与 DatePicker 并存: 快捷走 chips, 精确日期走日历。 + */ +export function DateShortcuts({ value, onChange, options, base }: Props) { + return ( +
| 标的 | +数量(参考) | +成本价 | +现价 | +盈亏% | +止盈% | +止损% | +买入日期 | +剩余天数 | +到期提醒 | ++ |
|---|---|---|---|---|---|---|---|---|---|---|
| + + | +{lot.qty} | +{lot.cost_price} | +{lastPrices[lot.symbol] != null ? fmtPrice(lastPrices[lot.symbol]) : '—'} | +{lot.target_pct > 0 ? `${lot.target_pct}%` : '—'} | +{lot.stop_pct > 0 ? `${lot.stop_pct}%` : '—'} | +{lot.buy_date || '—'} | +
+ {lot.remind_date ? (
+
+ |
+
+
+
+ {confirmId === lot.id ? (
+
+ ) : (
+
+ )}
+
+ |
+