From bd2fb5cc33e9e405f4cd440cffc63c2ae9260c02 Mon Sep 17 00:00:00 2001 From: OG T Date: Fri, 19 Jun 2026 00:14:07 +0800 Subject: [PATCH] feat: refresh recommendation calibration from settled performance --- .../app/analytics/daily_card_generator.py | 28 +++- platform/backend/app/main.py | 134 +++++++++++++++++- 2 files changed, 158 insertions(+), 4 deletions(-) diff --git a/platform/backend/app/analytics/daily_card_generator.py b/platform/backend/app/analytics/daily_card_generator.py index ec69547..961363e 100644 --- a/platform/backend/app/analytics/daily_card_generator.py +++ b/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: - return RECENT_MARKET_CALIBRATION[normalized_market] + calibration_source = _RUNTIME_MARKET_CALIBRATION or RECENT_MARKET_CALIBRATION + 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 diff --git a/platform/backend/app/main.py b/platform/backend/app/main.py index 283d0d0..d8cb5e7 100644 --- a/platform/backend/app/main.py +++ b/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()