diff --git a/frontend/lib/dashboard-types.ts b/frontend/lib/dashboard-types.ts
index 6400f853..dfa4da25 100644
--- a/frontend/lib/dashboard-types.ts
+++ b/frontend/lib/dashboard-types.ts
@@ -527,6 +527,22 @@ export interface HistoryPoint {
mu?: number | null;
mgm?: number | null;
forecasts?: Record;
+ model_reference?: {
+ available?: boolean;
+ truth_layer?: string | null;
+ reference_layer?: string | null;
+ deb?: {
+ value?: number | null;
+ error?: number | null;
+ };
+ models?: Array<{
+ model?: string | null;
+ value?: number | null;
+ error?: number | null;
+ participates_in_deb?: boolean;
+ }>;
+ model_count?: number | null;
+ };
settlement_source?: string | null;
settlement_station_code?: string | null;
settlement_station_label?: string | null;
diff --git a/src/bot/runtime_coordinator.py b/src/bot/runtime_coordinator.py
index 4a8c2642..3dfb7b3f 100644
--- a/src/bot/runtime_coordinator.py
+++ b/src/bot/runtime_coordinator.py
@@ -162,10 +162,24 @@ class StartupCoordinator:
"cities_count": cities_count,
"chat_targets": len(chat_ids),
"focus_digest_enabled": _env_bool("TELEGRAM_MARKET_FOCUS_DIGEST_ENABLED", True),
- "focus_digest_hours": str(
- os.getenv("TELEGRAM_MARKET_FOCUS_DIGEST_HOURS") or "11,18"
- ).strip(),
- "schedule_basis": "server_local_time",
+ "focus_digest_interval_sec": max(
+ 300,
+ _env_int("TELEGRAM_MARKET_FOCUS_DIGEST_INTERVAL_SEC", 1800),
+ ),
+ "focus_push_start_local_hour": min(
+ 23,
+ max(0, _env_int("TELEGRAM_MARKET_FOCUS_PUSH_START_LOCAL_HOUR", 8)),
+ ),
+ "focus_push_before_peak_min": max(
+ 0,
+ _env_int("TELEGRAM_MARKET_FOCUS_PUSH_BEFORE_PEAK_MIN", 480),
+ ),
+ "focus_push_after_peak_min": max(
+ 0,
+ _env_int("TELEGRAM_MARKET_FOCUS_PUSH_AFTER_PEAK_MIN", 300),
+ ),
+ "schedule_basis": "city_local_time_for_push_window",
+ "daily_signal_cap": "none",
}
validation_error = None if chat_ids else "missing_TELEGRAM_CHAT_IDS"
return self._start_with_validation(
diff --git a/src/utils/telegram_push.py b/src/utils/telegram_push.py
index 17a5e565..12a0915e 100644
--- a/src/utils/telegram_push.py
+++ b/src/utils/telegram_push.py
@@ -317,6 +317,32 @@ def _format_interval_brief(seconds: int) -> str:
return f"{total}秒"
+def _focus_push_window_config() -> Dict[str, int]:
+ return {
+ "start_local_hour": min(
+ 23,
+ max(0, _env_int("TELEGRAM_MARKET_FOCUS_PUSH_START_LOCAL_HOUR", 8)),
+ ),
+ "before_peak_min": max(
+ 0,
+ _env_int("TELEGRAM_MARKET_FOCUS_PUSH_BEFORE_PEAK_MIN", 480),
+ ),
+ "after_peak_min": max(
+ 0,
+ _env_int("TELEGRAM_MARKET_FOCUS_PUSH_AFTER_PEAK_MIN", 300),
+ ),
+ }
+
+
+def _focus_push_window_label() -> str:
+ config = _focus_push_window_config()
+ return (
+ f"当地 {config['start_local_hour']:02d}:00 后，"
+ f"峰值前 {_format_minutes_window(config['before_peak_min'])} / "
+ f"峰值后 {_format_minutes_window(config['after_peak_min'])} 内"
+ )
+
+
def _local_peak_context(alert_payload: Dict[str, Any]) -> Dict[str, Any]:
evidence = alert_payload.get("evidence") or {}
generated_local_time = str(evidence.get("generated_local_time") or "").strip()
@@ -456,15 +482,16 @@ def _shortlist_focus_payloads(
if not _market_price_cap_ok(item, require_actionable_quote=True):
continue
if for_push:
+ window_config = _focus_push_window_config()
peak_context = _local_peak_context(item)
local_min = _minute_of_day(peak_context.get("local_time"))
minutes_to_peak = peak_context.get("minutes_to_peak")
- if local_min is not None and local_min < 9 * 60:
+ if local_min is not None and local_min < window_config["start_local_hour"] * 60:
continue
- if isinstance(minutes_to_peak, int | float):
- if minutes_to_peak > 360:
+ if isinstance(minutes_to_peak, (int, float)):
+ if minutes_to_peak > window_config["before_peak_min"]:
continue
- if minutes_to_peak < -90:
+ if minutes_to_peak < -window_config["after_peak_min"]:
continue
shortlisted.append(item)
if len(shortlisted) >= top_n:
@@ -542,6 +569,8 @@ def _build_focus_digest_message(
frequency_parts = [
f"后台扫描：约每{scan_interval}一次",
f"主动推送：约每{digest_interval}一次",
+ f"观察窗口：{_focus_push_window_label()}",
+ "没有每日信号次数上限",
]
lines.append("更新频率：" + "；".join(frequency_parts))
return "\n".join(lines).strip()
@@ -576,17 +605,20 @@ def _maybe_send_focus_digest(
if bool((item.get("market_snapshot") or {}).get("available"))
and _market_price_cap_ok(item, require_actionable_quote=True)
)
- shortlisted = _shortlist_focus_payloads(payloads, top_n=top_n)
+ shortlisted = _shortlist_focus_payloads(payloads, top_n=top_n, for_push=True)
+ reference_shortlisted = _shortlist_focus_payloads(payloads, top_n=top_n, for_push=False)
high_priority_count = sum(1 for item in shortlisted if _market_monitor_score(item) >= 72)
logger.info(
- "market focus digest evaluate payloads={} available={} actionable={} shortlisted={} high_priority={} interval_sec={} top_n={}",
+ "market focus digest evaluate payloads={} available={} actionable={} shortlisted={} reference_shortlisted={} high_priority={} interval_sec={} top_n={} push_window={}",
len(payloads),
available_count,
actionable_count,
len(shortlisted),
+ len(reference_shortlisted),
high_priority_count,
digest_interval_sec,
top_n,
+ _focus_push_window_label(),
)
if last_digest_ts and now_ts - last_digest_ts < digest_interval_sec:
logger.info(
diff --git a/tests/test_market_alert_engine.py b/tests/test_market_alert_engine.py
index 9112a08c..9a90d22a 100644
--- a/tests/test_market_alert_engine.py
+++ b/tests/test_market_alert_engine.py
@@ -1,5 +1,9 @@
from src.analysis.market_alert_engine import build_trading_alerts
-from src.utils.telegram_push import build_market_monitor_digest
+from src.utils.telegram_push import (
+ _build_focus_digest_message,
+ _shortlist_focus_payloads,
+ build_market_monitor_digest,
+)
def _sample_weather_payload():
@@ -195,3 +199,62 @@ def test_market_monitor_digest_skips_non_tradable_market(monkeypatch):
digest = build_market_monitor_digest({}, slot_label="当前市场概览")
assert digest == "ℹ️ 当前没有可用的市场监控摘要。"
+
+
+def _focus_payload(*, local_time: str = "14:00", peak_time: str = "12:00"):
+ return {
+ "city": "madrid",
+ "severity": "medium",
+ "trigger_count": 1,
+ "rules": {},
+ "market_snapshot": {
+ "available": True,
+ "forecast_bucket": {
+ "label": "30°C",
+ "value": 30,
+ "yes_buy": 0.18,
+ "market_url": "https://example.com/market",
+ },
+ "market_url": "https://example.com/market",
+ "confidence": "medium",
+ "edge_percent": 5.0,
+ },
+ "evidence": {
+ "generated_local_time": local_time,
+ "inputs": {
+ "current_temp": 29.0,
+ "deb_prediction": 30.2,
+ },
+ "trigger_summary": {
+ "suppression_snapshot": {
+ "max_temp_time": peak_time,
+ },
+ },
+ },
+ }
+
+
+def test_focus_digest_push_window_does_not_cut_off_after_2pm_spanish_time(monkeypatch):
+ monkeypatch.delenv("TELEGRAM_MARKET_FOCUS_PUSH_AFTER_PEAK_MIN", raising=False)
+
+ shortlisted = _shortlist_focus_payloads(
+ [_focus_payload(local_time="14:00", peak_time="12:00")],
+ top_n=5,
+ for_push=True,
+ )
+
+ assert len(shortlisted) == 1
+
+
+def test_focus_digest_footer_explains_no_daily_signal_cap(monkeypatch):
+ monkeypatch.setenv("TELEGRAM_ALERT_PUSH_INTERVAL_SEC", "300")
+ monkeypatch.setenv("TELEGRAM_MARKET_FOCUS_DIGEST_INTERVAL_SEC", "1800")
+
+ message = _build_focus_digest_message(
+ [_focus_payload(local_time="14:00", peak_time="12:00")],
+ slot_label="白天关注",
+ top_n=5,
+ )
+
+ assert "没有每日信号次数上限" in message
+ assert "观察窗口" in message
diff --git a/web/routes.py b/web/routes.py
index 99798202..5ee19d06 100644
--- a/web/routes.py
+++ b/web/routes.py
@@ -203,6 +203,52 @@ def _refresh_city_market_cache(city: str, force_refresh: bool = False) -> dict:
return payload
+def _build_history_model_reference(
+ *,
+ forecasts: dict,
+ actual: object,
+ deb: object,
+) -> dict:
+ """Expose the archived model snapshot as reference evidence, not truth."""
+ actual_value = _sf(actual)
+ deb_value = _sf(deb)
+ entries = []
+ for model_name, model_value in (forecasts or {}).items():
+ if _is_excluded_model_name(str(model_name)):
+ continue
+ value = _sf(model_value)
+ if value is None:
+ continue
+ error = abs(value - actual_value) if actual_value is not None else None
+ entries.append(
+ {
+ "model": str(model_name),
+ "value": round(value, 1),
+ "error": round(error, 1) if error is not None else None,
+ "participates_in_deb": True,
+ }
+ )
+ entries.sort(
+ key=lambda row: (
+ row["error"] is None,
+ row["error"] if row["error"] is not None else 999,
+ row["model"],
+ )
+ )
+ deb_error = abs(deb_value - actual_value) if deb_value is not None and actual_value is not None else None
+ return {
+ "available": bool(entries),
+ "truth_layer": "settlement_actual",
+ "reference_layer": "archived_model_snapshot",
+ "deb": {
+ "value": round(deb_value, 1) if deb_value is not None else None,
+ "error": round(deb_error, 1) if deb_error is not None else None,
+ },
+ "models": entries,
+ "model_count": len(entries),
+ }
+
+
def _build_city_history_payload(city: str, include_records: bool = False) -> dict:
source = str(CITIES.get(city, {}).get("settlement_source") or "metar").strip().lower()
truth_rows = _truth_record_repo.load_city(city)
@@ -276,6 +322,11 @@ def _build_city_history_payload(city: str, include_records: bool = False) -> dic
forecasts,
snapshots,
)
+ model_reference = _build_history_model_reference(
+ forecasts=forecasts,
+ actual=act,
+ deb=deb,
+ )
mgm = forecasts.get("MGM")
out.append(
{
@@ -285,6 +336,7 @@ def _build_city_history_payload(city: str, include_records: bool = False) -> dic
"mu": float(mu) if mu is not None else None,
"mgm": float(mgm) if mgm is not None else None,
"forecasts": forecasts,
+ "model_reference": model_reference,
"settlement_source": rec.get("settlement_source"),
"settlement_station_code": rec.get("settlement_station_code"),
"settlement_station_label": rec.get("settlement_station_label"),