feat: refresh recommendation calibration from settled performance

platform/backend/app/analytics/daily_card_generator.py

@@ -135,6 +135,27 @@ RECENT_MARKET_CALIBRATION: dict[str, dict[str, Any]] = {
   },
 }
 
+_RUNTIME_MARKET_CALIBRATION: dict[str, dict[str, Any]] | None = None
+
+
+def update_runtime_market_calibration(market_calibration: dict[str, dict[str, Any]] | None) -> None:
+  """Update in-process market calibration from settled recommendation performance."""
+
+  global _RUNTIME_MARKET_CALIBRATION
+  if not market_calibration:
+    _RUNTIME_MARKET_CALIBRATION = None
+    return
+
+  cleaned: dict[str, dict[str, Any]] = {}
+  for market_type, calibration in market_calibration.items():
+    if not isinstance(calibration, dict):
+      continue
+    normalized_market = str(market_type or '').strip()
+    if not normalized_market:
+      continue
+    cleaned[normalized_market] = dict(calibration)
+  _RUNTIME_MARKET_CALIBRATION = cleaned or None
+
 
 def _safe_float(value: Any, default: float = 0.0) -> float:
   try:
@@ -546,10 +567,11 @@ def _recent_market_calibration(market_type: str) -> dict[str, Any] | None:
   """Return recent post-match market calibration for public recommendation safety."""
 
   normalized_market = str(market_type or '').strip()
-  if normalized_market in RECENT_MARKET_CALIBRATION:
+  calibration_source = _RUNTIME_MARKET_CALIBRATION or RECENT_MARKET_CALIBRATION
-    return RECENT_MARKET_CALIBRATION[normalized_market]
+  if normalized_market in calibration_source:
+    return calibration_source[normalized_market]
   if normalized_market.startswith('大小球 '):
-    return RECENT_MARKET_CALIBRATION.get(normalized_market)
+    return calibration_source.get(normalized_market)
   return None
 
 

platform/backend/app/main.py

@@ -20,7 +20,11 @@ from pydantic import BaseModel, Field
 from redis.asyncio import Redis
 from .db.base import SessionFactory
 from .db.models import Bookmaker, DailyRecommendationSnapshot, Match, MatchStatus, OddsHistory, SmartMoneyFlow, Team, Venue
-from .analytics.daily_card_generator import generate_daily_card, recalibrate_daily_card_confidence_payload
+from .analytics.daily_card_generator import (
+  generate_daily_card,
+  recalibrate_daily_card_confidence_payload,
+  update_runtime_market_calibration,
+)
 from .analytics.localization import (
   localize_city,
   localize_country,
@@ -2229,6 +2233,107 @@ def _performance_actions(hit_rate: float, buckets: list[dict[str, Any]], items:
   return actions
 
 
+RECOMMENDATION_CALIBRATION_CACHE: dict[str, Any] = {
+  'expires_at': None,
+  'payload': None,
+}
+RECOMMENDATION_CALIBRATION_TTL_SECONDS = 600
+
+
+def _bucket_value(bucket: Any, key: str, default: Any = 0) -> Any:
+  if isinstance(bucket, dict):
+    return bucket.get(key, default)
+  return getattr(bucket, key, default)
+
+
+def _build_runtime_market_calibration(buckets: list[Any], days_back: int) -> dict[str, dict[str, Any]]:
+  calibration: dict[str, dict[str, Any]] = {}
+
+  for bucket in buckets:
+    market_type = str(_bucket_value(bucket, 'market_type', '') or '').strip()
+    if not market_type:
+      continue
+    settled_count = int(_bucket_value(bucket, 'settled_count', 0) or 0)
+    hit_count = int(_bucket_value(bucket, 'hit_count', 0) or 0)
+    miss_count = int(_bucket_value(bucket, 'miss_count', 0) or 0)
+    denominator = hit_count + miss_count
+    if denominator <= 0:
+      continue
+    hit_rate = round((hit_count / denominator) * 100, 2)
+
+    if hit_rate < 20.0 and settled_count >= 4:
+      severity = 'severe'
+      confidence_penalty = 10.0
+      stake_multiplier = 0.55
+      min_ev_boost = 5.0
+      min_win_prob_boost = 0.04
+      action_note = '先降為保守監控，不當核心下注'
+    elif hit_rate < 45.0 and settled_count >= 4:
+      severity = 'caution'
+      confidence_penalty = round(min(8.0, 3.0 + ((45.0 - hit_rate) / 4.0)), 1)
+      stake_multiplier = 0.70
+      min_ev_boost = 3.0
+      min_win_prob_boost = 0.02
+      action_note = '提高進場門檻並縮小注碼'
+    elif hit_rate >= 65.0 and settled_count >= 5:
+      severity = 'stable'
+      confidence_penalty = 0.0
+      stake_multiplier = 1.0
+      min_ev_boost = 0.0
+      min_win_prob_boost = 0.0
+      action_note = '暫列較穩玩法，但不自動加碼'
+    else:
+      continue
+
+    if market_type == '正確比分':
+      confidence_penalty = max(confidence_penalty, 11.0)
+      stake_multiplier = min(stake_multiplier, 0.42)
+      min_ev_boost = max(min_ev_boost, 8.0)
+      min_win_prob_boost = max(min_win_prob_boost, 0.03)
+    elif market_type in {'跨場串關', '同場串關'}:
+      confidence_penalty = max(confidence_penalty, 6.5)
+      stake_multiplier = min(stake_multiplier, 0.68)
+      min_ev_boost = max(min_ev_boost, 4.0)
+      min_win_prob_boost = max(min_win_prob_boost, 0.02)
+    elif market_type in {'勝平負', '大小球 2.5'} and severity == 'severe':
+      confidence_penalty = max(confidence_penalty, 9.0)
+      stake_multiplier = min(stake_multiplier, 0.58)
+      min_ev_boost = max(min_ev_boost, 5.0)
+      min_win_prob_boost = max(min_win_prob_boost, 0.035)
+
+    calibration[market_type] = {
+      'settled_count': settled_count,
+      'hit_rate_percent': hit_rate,
+      'confidence_penalty': confidence_penalty,
+      'stake_multiplier': stake_multiplier,
+      'min_ev_boost': min_ev_boost,
+      'min_win_prob_boost': min_win_prob_boost,
+      'severity': severity,
+      'note': (
+        f'近 {days_back} 天{market_type} {settled_count} 筆可判定、'
+        f'命中率 {hit_rate:.2f}%，系統已自動{action_note}。'
+      ),
+    }
+
+  return calibration
+
+
+async def _refresh_runtime_recommendation_calibration(days_back: int = 7) -> dict[str, dict[str, Any]]:
+  now = datetime.now(timezone.utc)
+  expires_at = RECOMMENDATION_CALIBRATION_CACHE.get('expires_at')
+  cached_payload = RECOMMENDATION_CALIBRATION_CACHE.get('payload')
+  if isinstance(expires_at, datetime) and expires_at > now and isinstance(cached_payload, dict):
+    update_runtime_market_calibration(cached_payload)
+    return cached_payload
+
+  performance = await _build_recommendation_performance(days_back)
+  calibration = _build_runtime_market_calibration(list(performance.by_market_type), days_back)
+  update_runtime_market_calibration(calibration)
+  RECOMMENDATION_CALIBRATION_CACHE['payload'] = calibration
+  RECOMMENDATION_CALIBRATION_CACHE['expires_at'] = now + timedelta(seconds=RECOMMENDATION_CALIBRATION_TTL_SECONDS)
+  return calibration
+
+
 async def _build_recommendation_performance(days_back: int) -> RecommendationPerformanceResponse:
   match_payload, result_lookup = await _query_finished_recommendation_snapshots(days_back)
   generated_at = datetime.now(timezone.utc).isoformat()
@@ -2351,6 +2456,12 @@ async def _build_recommendation_performance(days_back: int) -> RecommendationPer
     f'其中 {settled_count} 組可自動判定，命中 {hit_count} 組、未中 {miss_count} 組、退回 {push_count} 組，'
     f'命中率 {hit_rate:.2f}%。'
   )
+  runtime_calibration = _build_runtime_market_calibration(buckets, days_back)
+  update_runtime_market_calibration(runtime_calibration)
+  RECOMMENDATION_CALIBRATION_CACHE['payload'] = runtime_calibration
+  RECOMMENDATION_CALIBRATION_CACHE['expires_at'] = datetime.now(timezone.utc) + timedelta(
+    seconds=RECOMMENDATION_CALIBRATION_TTL_SECONDS
+  )
 
   return RecommendationPerformanceResponse(
     generated_at=generated_at,
@@ -4111,6 +4222,10 @@ async def daily_card_calendar_status_route() -> dict[str, Any]:
 @app.get('/analytics/daily-card/{target_date}', response_model=DailyCardResponse)
 async def generate_daily_card_route(target_date: str) -> DailyCardResponse:
   target_day = _to_date(target_date)
+  try:
+    await _refresh_runtime_recommendation_calibration(7)
+  except Exception:
+    pass
   snapshot_payload = await _read_daily_recommendation_snapshot_payload(target_date) if target_day <= _taipei_today_date() else None
   if target_day < _taipei_today_date() and snapshot_payload:
     return DailyCardResponse(**snapshot_payload)
@@ -4132,6 +4247,23 @@ async def recommendation_performance_route(days_back: int = 7) -> Recommendation
   return await _build_recommendation_performance(normalized_days_back)
 
 
+@app.get('/analytics/recommendation-calibration')
+async def recommendation_calibration_route(days_back: int = 7) -> dict[str, Any]:
+  normalized_days_back = max(1, min(days_back, 30))
+  calibration = await _refresh_runtime_recommendation_calibration(normalized_days_back)
+  return {
+    'generated_at': datetime.now(timezone.utc).isoformat(),
+    'days_back': normalized_days_back,
+    'market_count': len(calibration),
+    'cache_ttl_seconds': RECOMMENDATION_CALIBRATION_TTL_SECONDS,
+    'calibration': calibration,
+    'methodology_note': (
+      '系統會用近端已完賽推薦績效，把低命中玩法提高 EV/勝率門檻並縮小新台幣上限；'
+      '高命中玩法只標示較穩，不會因短期樣本直接加碼。'
+    ),
+  }
+
+
 @app.get('/analytics/agent-verification', response_model=AgentVerificationResponse)
 async def agent_verification_route() -> AgentVerificationResponse:
   return await _build_agent_verification()