From 7799c1e8977e5167554c5d110ea4f856e5b2b57b Mon Sep 17 00:00:00 2001 From: OG T Date: Thu, 18 Jun 2026 18:48:27 +0800 Subject: [PATCH] fix: calibrate betting recommendations from settled outcomes --- .../app/analytics/daily_card_generator.py | 226 ++++++++++++++++-- 1 file changed, 207 insertions(+), 19 deletions(-) diff --git a/platform/backend/app/analytics/daily_card_generator.py b/platform/backend/app/analytics/daily_card_generator.py index e1e0b08..fb55e45 100644 --- a/platform/backend/app/analytics/daily_card_generator.py +++ b/platform/backend/app/analytics/daily_card_generator.py @@ -52,6 +52,89 @@ TEAM_ALIASES = { 'drcongo': '剛果民主共和國', } +RECENT_MARKET_CALIBRATION: dict[str, dict[str, Any]] = { + '同場串關': { + 'settled_count': 10, + 'hit_rate_percent': 30.0, + 'confidence_penalty': 6.5, + 'stake_multiplier': 0.68, + 'min_ev_boost': 4.0, + 'min_win_prob_boost': 0.02, + 'severity': 'caution', + 'note': '近 7 天同場串關 10 筆可判定、命中率 30.00%，已提高進場門檻並縮小注碼。', + }, + '正確比分': { + 'settled_count': 7, + 'hit_rate_percent': 0.0, + 'confidence_penalty': 11.0, + 'stake_multiplier': 0.42, + 'min_ev_boost': 8.0, + 'min_win_prob_boost': 0.03, + 'severity': 'severe', + 'note': '近 7 天正確比分 7 筆可判定仍未命中，僅保留超高賠小倉監控。', + }, + '大小球 3.5': { + 'settled_count': 6, + 'hit_rate_percent': 33.33, + 'confidence_penalty': 5.5, + 'stake_multiplier': 0.72, + 'min_ev_boost': 3.0, + 'min_win_prob_boost': 0.015, + 'severity': 'caution', + 'note': '近 7 天大小球 3.5 命中率 33.33%，進場需要更明顯的賠率優勢。', + }, + '勝平負': { + 'settled_count': 6, + 'hit_rate_percent': 0.0, + 'confidence_penalty': 10.0, + 'stake_multiplier': 0.55, + 'min_ev_boost': 5.0, + 'min_win_prob_boost': 0.04, + 'severity': 'severe', + 'note': '近 7 天勝平負 6 筆可判定仍未命中，先降為保守監控，不當核心下注。', + }, + '跨場串關': { + 'settled_count': 5, + 'hit_rate_percent': 20.0, + 'confidence_penalty': 7.0, + 'stake_multiplier': 0.62, + 'min_ev_boost': 4.5, + 'min_win_prob_boost': 0.025, + 'severity': 'caution', + 'note': '近 7 天跨場串關命中率 20.00%，串關只作輔助，不放大倉位。', + }, + '大小球 2.5': { + 'settled_count': 4, + 'hit_rate_percent': 0.0, + 'confidence_penalty': 9.0, + 'stake_multiplier': 0.58, + 'min_ev_boost': 5.0, + 'min_win_prob_boost': 0.035, + 'severity': 'severe', + 'note': '近 7 天大小球 2.5 四筆全未命中，需等更好的盤口差距才進場。', + }, + '雙重機會': { + 'settled_count': 7, + 'hit_rate_percent': 71.43, + 'confidence_penalty': 0.0, + 'stake_multiplier': 1.0, + 'min_ev_boost': 0.0, + 'min_win_prob_boost': 0.0, + 'severity': 'stable', + 'note': '近 7 天雙重機會命中率 71.43%，暫列較穩玩法，但仍不額外加碼。', + }, + '大小球 1.5': { + 'settled_count': 5, + 'hit_rate_percent': 100.0, + 'confidence_penalty': 0.0, + 'stake_multiplier': 1.0, + 'min_ev_boost': 0.0, + 'min_win_prob_boost': 0.0, + 'severity': 'stable', + 'note': '近 7 天大小球 1.5 五筆全命中，暫保留原門檻；樣本仍小，不自動加碼。', + }, +} + def _safe_float(value: Any, default: float = 0.0) -> float: try: @@ -459,6 +542,98 @@ def _finalize_confidence_score( return round(_clamp(adjusted, floor, ceiling), 1) +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] + if normalized_market.startswith('大小球 '): + return RECENT_MARKET_CALIBRATION.get(normalized_market) + return None + + +def _calibrated_minimums(market_type: str, min_win_prob: float, min_ev: float) -> tuple[float, float]: + calibration = _recent_market_calibration(market_type) + if not calibration: + return min_win_prob, min_ev + return ( + min_win_prob + _safe_float(calibration.get('min_win_prob_boost'), 0.0), + min_ev + _safe_float(calibration.get('min_ev_boost'), 0.0), + ) + + +def _calibration_marker(market_type: str) -> str: + return f'近7天玩法校準：{market_type}' + + +def _apply_recent_market_calibration(item: dict[str, Any]) -> dict[str, Any]: + """Apply settled-result feedback to a generated pick without changing its legs.""" + + market_type = str(item.get('market_type') or '') + calibration = _recent_market_calibration(market_type) + if not calibration: + return item + + marker = _calibration_marker(market_type) + data_checks = list(item.get('data_checks') or []) + if marker in data_checks: + return item + + adjusted = dict(item) + factors = list(adjusted.get('confidence_factors') or []) + note = str(calibration.get('note') or '') + settled_count = int(_safe_float(calibration.get('settled_count'), 0)) + hit_rate = _safe_float(calibration.get('hit_rate_percent'), 0.0) + severity = str(calibration.get('severity') or 'neutral') + confidence_penalty = _safe_float(calibration.get('confidence_penalty'), 0.0) + stake_multiplier = _safe_float(calibration.get('stake_multiplier'), 1.0) + + current_confidence = _safe_float(adjusted.get('confidence_score'), 0.0) + if confidence_penalty > 0: + adjusted['confidence_score'] = round(_clamp(current_confidence - confidence_penalty, 0.0, current_confidence), 1) + + current_stake = _safe_float(adjusted.get('stake_units'), 0.0) + if current_stake > 0 and stake_multiplier < 1.0: + next_stake = round(_clamp(current_stake * stake_multiplier, 0.05, current_stake), 2) + adjusted['stake_units'] = next_stake + adjusted['stake_amount_twd'] = _stake_amount_twd(next_stake) + + if severity == 'severe': + recommendation = str(adjusted.get('recommendation') or '') + if recommendation == 'SAFE_SINGLE': + adjusted['recommendation'] = 'WATCHLIST_SINGLE' + elif recommendation == 'HIGH_RISK_SINGLE': + adjusted['recommendation'] = 'WATCHLIST_HIGH_RISK' + elif recommendation == 'SAFE_PARLAY': + adjusted['recommendation'] = 'WATCHLIST_PARLAY' + if str(adjusted.get('risk_level') or '') == 'core': + adjusted['risk_level'] = 'speculative' + + adjusted['confidence_band'] = _confidence_band( + _safe_float(adjusted.get('confidence_score'), 0.0), + bool(adjusted.get('has_market_odds')), + str(adjusted.get('risk_level') or 'core'), + ) + + if note and note not in factors: + factors.append(note) + factors.append(f'賽後校準樣本 {settled_count} 筆，命中率 {hit_rate:.2f}%') + adjusted['confidence_factors'] = factors[:9] + adjusted['data_checks'] = data_checks + [ + marker, + f'近7天樣本 {settled_count} 筆', + f'近7天命中率 {hit_rate:.2f}%', + '已套用賽後校準觀察' if severity == 'stable' else '已套用賽後校準降權', + ] + + rationale = str(adjusted.get('rationale') or '') + if note and note not in rationale: + adjusted['rationale'] = f'{rationale} 賽後校準：{note}' + + return adjusted + + def _market_tier_from_item(item: dict[str, Any]) -> str: market_type = str(item.get('market_type') or '') @@ -530,6 +705,7 @@ def recalibrate_daily_card_confidence_payload(card_payload: dict[str, Any]) -> d if '信心分數已套用新版資料品質校準' not in factors: factors.append('信心分數已套用新版資料品質校準') item['confidence_factors'] = factors[:8] + item = _apply_recent_market_calibration(item) except Exception: pass next_group.append(item) @@ -752,7 +928,8 @@ def _build_pick( ev = _ev_percent(true_prob, odds) edge = _edge_percent(true_prob, odds) - if true_prob < min_win_prob or ev < min_ev or edge < min_edge: + effective_min_win_prob, effective_min_ev = _calibrated_minimums(market_type, min_win_prob, min_ev) + if true_prob < effective_min_win_prob or ev < effective_min_ev or edge < min_edge: return None stake = _stake_units(true_prob, ev, risk_level, stake_cap) @@ -811,7 +988,7 @@ def _build_pick( market_overround=market_overround, ) - return { + return _apply_recent_market_calibration({ 'match_id': match_id, 'match_label': match_label, 'market_type': market_type, @@ -839,7 +1016,7 @@ def _build_pick( f'{fair_note}{quality_note}{longshot_note}建議上限約新台幣 {_stake_amount_twd(stake):,} 元（{stake:.2f}u），且需等待最新傷停、先發與盤口刷新一致後才進入檢驗單。' ), 'data_checks': enriched_checks, - } + }) def _minimum_value_odds(true_prob: float, min_ev_percent: float) -> float: @@ -870,19 +1047,20 @@ def _build_conditional_pick( ) -> dict[str, Any] | None: """尚無即時盤口時，產生可執行的預掛進場條件，而不是假裝已有市場賠率。""" - if true_prob < min_win_prob: + effective_min_win_prob, effective_min_ev = _calibrated_minimums(market_type, min_win_prob, min_ev) + if true_prob < effective_min_win_prob: return None - target_odds = _minimum_value_odds(true_prob, min_ev) + target_odds = _minimum_value_odds(true_prob, effective_min_ev) if target_odds <= 1 or target_odds > max_target_odds: return None implied = (1.0 / target_odds) * 100 edge = (true_prob * 100) - implied - stake = _stake_units(true_prob, min_ev, risk_level, stake_cap) + stake = _stake_units(true_prob, effective_min_ev, risk_level, stake_cap) confidence = _confidence_score( true_prob, - min_ev, + effective_min_ev, edge, market_tier, decimal_odds=target_odds, @@ -907,28 +1085,28 @@ def _build_conditional_pick( odds_source_kind=odds_source_kind, target_odds=target_odds, true_prob=true_prob, - ev_percent=min_ev, + ev_percent=effective_min_ev, edge_percent=edge, ) quality_note = _quality_note(data_quality, has_market_odds) confidence_band = _confidence_band(confidence, has_market_odds, risk_level) confidence_factors = _confidence_factors( true_prob=true_prob, - ev_percent=min_ev, + ev_percent=effective_min_ev, edge_percent=edge, data_quality=data_quality, has_market_odds=has_market_odds, market_tier=market_tier, ) - return { + return _apply_recent_market_calibration({ 'match_id': match_id, 'match_label': match_label, 'market_type': market_type, 'selection': f'{selection}｜預掛條件', 'target_odds': round(target_odds, 2), 'win_prob': round(true_prob * 100, 2), - 'ev_percent': round(min_ev, 2), + 'ev_percent': round(effective_min_ev, 2), 'stake_units': stake, 'stake_amount_twd': _stake_amount_twd(stake), 'unit_size_twd': DEFAULT_UNIT_STAKE_TWD, @@ -945,11 +1123,11 @@ def _build_conditional_pick( 'edge_percent': round(edge, 2), 'rationale': ( f'此場尚未取得可用即時盤口，因此先給「預掛進場條件」。模型勝率 {true_prob * 100:.2f}%，' - f'只有當平台賠率達到 {target_odds:.2f} 以上，才符合至少 {min_ev:.2f}% 的正期望值門檻；' + f'只有當平台賠率達到 {target_odds:.2f} 以上，才符合至少 {effective_min_ev:.2f}% 的正期望值門檻；' f'若實際賠率低於此數字，直接跳過，不追價。建議監控上限約新台幣 {_stake_amount_twd(stake):,} 元（{stake:.2f}u）。{quality_note}' ), 'data_checks': data_checks + [odds_source_label, '尚待盤口確認', '預掛最低賠率', '不得低於目標賠率下注'] + _quality_checks(data_quality, has_market_odds), - } + }) def _sort_picks(items: list[dict[str, Any]]) -> list[dict[str, Any]]: @@ -1016,6 +1194,7 @@ def _build_cross_match_parlays(safe_singles: list[dict[str, Any]]) -> list[dict[ combined_prob *= discount min_prob = 0.22 if size == 2 else 0.12 min_ev = 2.5 if size == 2 else 5.0 + min_prob, min_ev = _calibrated_minimums('跨場串關', min_prob, min_ev) required_combo_odds = _minimum_value_odds(combined_prob, min_ev) target_combo_odds = max(combined_odds, required_combo_odds) max_combo_odds = 8.5 if size == 2 else 18.0 @@ -1066,7 +1245,7 @@ def _build_cross_match_parlays(safe_singles: list[dict[str, Any]]) -> list[dict[ edge_percent=edge, ) parlays.append( - { + _apply_recent_market_calibration({ 'match_id': f'PARLAY-CROSS-MATCH-{size}-{len(parlays) + 1}', 'match_label': ' + '.join(item['match_label'] for item in legs), 'market_type': '跨場串關', @@ -1106,7 +1285,7 @@ def _build_cross_match_parlays(safe_singles: list[dict[str, Any]]) -> list[dict[ f'建議上限約新台幣 {_stake_amount_twd(stake_units):,} 元（{stake_units:.2f}u）。串關只作輔助倉位，不能取代單關主策略。' ), 'data_checks': ['單關門檻通過', combo_source_label, '跨場去相關', f'相關性 {discount:.2f} 折減', '串關總賠率門檻', '串關 EV 重新計算'] + _quality_checks(combo_data_quality, combo_has_market_odds), - } + }) ) return _sort_picks(parlays)[:6] @@ -1145,11 +1324,12 @@ def _build_same_game_parlays(picks_by_match: dict[str, list[dict[str, Any]]]) -> combo_odds = first['target_odds'] * second['target_odds'] combo_prob = (first['win_prob'] / 100) * (second['win_prob'] / 100) combo_prob *= 0.68 - required_combo_odds = _minimum_value_odds(combo_prob, 8.0) + min_prob, min_ev = _calibrated_minimums('同場串關', 0.16, 8.0) + required_combo_odds = _minimum_value_odds(combo_prob, min_ev) target_combo_odds = max(combo_odds, required_combo_odds) ev = _ev_percent(combo_prob, target_combo_odds) edge = _edge_percent(combo_prob, target_combo_odds) - if combo_prob < 0.16 or target_combo_odds > 12.0 or ev < 8.0 or edge <= 0: + if combo_prob < min_prob or target_combo_odds > 12.0 or ev < min_ev or edge <= 0: continue sgp_has_market_odds = False @@ -1192,7 +1372,7 @@ def _build_same_game_parlays(picks_by_match: dict[str, list[dict[str, Any]]]) -> ) sgps.append( - { + _apply_recent_market_calibration({ 'match_id': match_id, 'match_label': f"{first['match_label']}【同場】", 'market_type': '同場串關', @@ -1232,7 +1412,7 @@ def _build_same_game_parlays(picks_by_match: dict[str, list[dict[str, Any]]]) -> f'建議上限約新台幣 {_stake_amount_twd(sgp_stake_units):,} 元（{sgp_stake_units:.2f}u）。此類組合波動極大，只允許娛樂型小注，不可當核心下注。' ), 'data_checks': ['同場不同市場', sgp_source_label or '模型推算最低門檻', '相關性 0.68 折減', '預掛總賠率門檻', '高 EV 門檻', '小倉位上限'] + _quality_checks(sgp_data_quality, sgp_has_market_odds), - } + }) ) return _sort_picks(sgps)[:4] @@ -1598,6 +1778,13 @@ def generate_daily_card(target_date: str, matches: list[dict[str, Any]]) -> dict ) all_items = [*safe_singles, *high_risk_singles, *safe_parlays, *sgp_lotteries] + calibrated_count = sum( + 1 + for item in all_items + if any(str(check).startswith('近7天玩法校準') for check in (item.get('data_checks') or [])) + ) + if calibrated_count: + summary = f'{summary} 已套用近 7 天賽後玩法校準 {calibrated_count} 組，近期低命中玩法會自動提高門檻並縮小注碼。' live_market_count = sum(1 for item in all_items if bool(item.get('has_market_odds'))) pre_market_count = len(all_items) - live_market_count quality_counts: dict[str, int] = {} @@ -1616,6 +1803,7 @@ def generate_daily_card(target_date: str, matches: list[dict[str, Any]]) -> dict **quality_counts, 'live_market_count': live_market_count, 'pre_market_count': pre_market_count, + 'recent_market_calibrated_count': calibrated_count, }, 'execution_policy': ( '已有實盤盤口的候選可進入下注前檢查；尚未取得實盤者只能加入賠率監控，'