wooo/2026FIFAWorldCup
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit with 2026 World Cup Quant Platform core modules and CI/CD

Browse Source
This commit is contained in:
QuantBot
2026-06-13 23:18:18 +08:00
commit 073abf98c1
155 changed files with 19539 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
platform/backend/app/analytics/__init__.py Normal file
View File

@@ -0,0 +1,95 @@
"""量化分析模組匯總。"""
from .engine import calculate_value_bet, PoissonPredictor, adjust_away_defense_for_altitude
from .ev_calculator import calculate_expected_value
from .feature_engineering import MatchFeatureExtractor, MatchFeatureVector
from .kelly import KellyResult, calculate_kelly_fraction
from .ml_inference import XGBoostPredictor, XGBoostPrediction
from .player_props import (
PlayerPropsProfile,
PlayerPropsSimulationResult,
PropMetric,
evaluate_top_edge,
simulate_player_prop_probability,
)
from .ml_ensemble import (
FEATURE_COLUMNS,
EnsembleModelArtifact,
build_default_ensemble_artifact,
calculate_model_edges,
model_predict_probabilities,
normalize_feature_payload,
train_match_outcome_ensemble,
)
from .backtesting import BacktestTradeRecord, StrategyFilter, filter_trades, run_flat_stake_backtest
from .poisson_model import PoissonMatchPredictor
from .referee_analyzer import calculate_cards_ev
from .environment_model import adjust_team_strength_for_environment
from .referee_weather import MatchConditionSignal, evaluate_match_conditions
from .rlm import ReverseLineMovementAlert, evaluate_reverse_line_movement
from .proof_of_yield import LedgerSummary, ProofOfYieldStore, ProofYieldRecord, compute_clv, compute_pnl
from .player_props_sim import PlayerPropsDistribution, evaluate_prop_bet, simulate_player_stats
from .sgp_engine import calculate_joint_probability, find_sgp_value
from .portfolio_analyzer import analyze_user_leaks
from .hedging_calculator import calculate_hedge_amount
from .daily_card_generator import generate_daily_card
from .vig_remover import (
calculate_overround,
compare_bookmaker_true_prob,
prob_to_decimal_odds,
remove_margin_basic,
remove_margin_shin,
)
__all__ = [
'KellyResult',
'BacktestTradeRecord',
'PropMetric',
'calculate_expected_value',
'calculate_value_bet',
'calculate_kelly_fraction',
'evaluate_top_edge',
'PoissonPredictor',
'PlayerPropsProfile',
'PlayerPropsSimulationResult',
'PoissonMatchPredictor',
'adjust_away_defense_for_altitude',
'adjust_team_strength_for_environment',
'filter_trades',
'run_flat_stake_backtest',
'simulate_player_prop_probability',
'StrategyFilter',
'FEATURE_COLUMNS',
'build_default_ensemble_artifact',
'calculate_model_edges',
'model_predict_probabilities',
'normalize_feature_payload',
'train_match_outcome_ensemble',
'MatchConditionSignal',
'evaluate_match_conditions',
'ReverseLineMovementAlert',
'evaluate_reverse_line_movement',
'LedgerSummary',
'ProofOfYieldStore',
'ProofYieldRecord',
'compute_clv',
'compute_pnl',
'MatchFeatureExtractor',
'MatchFeatureVector',
'XGBoostPredictor',
'XGBoostPrediction',
'PlayerPropsDistribution',
'simulate_player_stats',
'evaluate_prop_bet',
'calculate_joint_probability',
'find_sgp_value',
'calculate_cards_ev',
'calculate_overround',
'remove_margin_basic',
'remove_margin_shin',
'prob_to_decimal_odds',
'compare_bookmaker_true_prob',
'analyze_user_leaks',
'calculate_hedge_amount',
'generate_daily_card',
]

181
platform/backend/app/analytics/backtesting.py Normal file
View File

@@ -0,0 +1,181 @@
"""自訂策略回測引擎。"""
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime
@dataclass(frozen=True)
class BacktestTradeRecord:
"""單筆策略投注歷史資料。"""
trade_id: str
settled_at: datetime
odds: float
is_win: bool
stake: float = 100.0
altitude_meters: int | None = None
handicap: float | None = None
weather: str | None = None
recent_form_win_rate: float | None = None
market_type: str = '1x2'
selection: str = 'home'
@dataclass(frozen=True)
class StrategyFilter:
"""回測條件（前端 JSON 可直接對映）。"""
weather: str | None = None
altitude_min_meters: int | None = None
altitude_max_meters: int | None = None
handicap_min: float | None = None
handicap_max: float | None = None
recent_win_rate_min: float | None = None
recent_win_rate_max: float | None = None
market_types: list[str] | None = None
start_at: datetime | None = None
end_at: datetime | None = None
def _match_filter(record: BacktestTradeRecord, condition: StrategyFilter) -> bool:
"""判斷單筆交易是否符合使用者條件。"""
if condition.weather and (record.weather or '').lower() != condition.weather.lower():
return False
if condition.altitude_min_meters is not None and (
record.altitude_meters is None or record.altitude_meters < condition.altitude_min_meters
):
return False
if condition.altitude_max_meters is not None and (
record.altitude_meters is None or record.altitude_meters > condition.altitude_max_meters
):
return False
if condition.handicap_min is not None and (
record.handicap is None or record.handicap < condition.handicap_min
):
return False
if condition.handicap_max is not None and (
record.handicap is None or record.handicap > condition.handicap_max
):
return False
if condition.recent_win_rate_min is not None and (
record.recent_form_win_rate is None or record.recent_form_win_rate < condition.recent_win_rate_min
):
return False
if condition.recent_win_rate_max is not None and (
record.recent_form_win_rate is None or record.recent_form_win_rate > condition.recent_win_rate_max
):
return False
if condition.market_types and record.market_type not in condition.market_types:
return False
if condition.start_at is not None and record.settled_at < condition.start_at:
return False
if condition.end_at is not None and record.settled_at > condition.end_at:
return False
return True
def filter_trades(
trades: list[BacktestTradeRecord],
condition: StrategyFilter,
) -> list[BacktestTradeRecord]:
"""回傳符合條件的策略明細子集合。"""
return [t for t in trades if _match_filter(t, condition)]
def compute_max_drawdown(equity_curve: list[float]) -> float:
"""計算最大回撤（百分比）。"""
if not equity_curve:
return 0.0
peak = equity_curve[0]
max_drawdown = 0.0
for value in equity_curve:
if value > peak:
peak = value
continue
drawdown = (peak - value) / peak if peak else 0.0
max_drawdown = max(max_drawdown, drawdown)
return round(max_drawdown * 100, 4)
def run_flat_stake_backtest(
trades: list[BacktestTradeRecord],
initial_capital: float = 10000,
) -> dict[str, float | int | list[dict[str, float | str]]]:
"""固定單注本金（Flat betting）回測。
回傳：
- trade_count：總注單數
- hit_count：中獎注數
- win_rate：中獎率
- final_capital：最終資金
- net_profit：淨利潤
- roi_percent：ROI
- max_drawdown_percent：最大回撤百分比
- equity_curve：資產曲線
"""
if initial_capital <= 0:
raise ValueError('initial_capital 必須大於 0')
if not trades:
return {
'trade_count': 0,
'hit_count': 0,
'win_rate': 0.0,
'final_capital': initial_capital,
'net_profit': 0.0,
'roi_percent': 0.0,
'max_drawdown_percent': 0.0,
'equity_curve': [{'ts': datetime.utcnow().isoformat() + 'Z', 'capital': initial_capital}],
}
# 確保輸入依賴的時序，回測才有金融合理性
ordered = sorted(trades, key=lambda row: row.settled_at)
equity = float(initial_capital)
equity_curve: list[dict[str, float | str]] = [
{'ts': ordered[0].settled_at.isoformat(), 'capital': equity},
]
hit = 0
total_stake = 0.0
for trade in ordered:
if trade.odds <= 1:
raise ValueError(f'賠率錯誤 trade={trade.trade_id}, odds={trade.odds}')
stake = trade.stake
profit = stake * (trade.odds - 1) if trade.is_win else -stake
equity += profit
total_stake += stake
if trade.is_win:
hit += 1
equity_curve.append({'ts': trade.settled_at.isoformat(), 'capital': equity})
if total_stake <= 0:
roi = 0.0
else:
roi = (equity - initial_capital) / total_stake * 100
win_rate = round(hit / len(ordered) * 100, 4) if ordered else 0.0
return {
'trade_count': len(ordered),
'hit_count': hit,
'win_rate': win_rate,
'final_capital': round(equity, 4),
'net_profit': round(equity - initial_capital, 4),
'roi_percent': round(roi, 4),
'max_drawdown_percent': compute_max_drawdown([float(point['capital']) for point in equity_curve]),
'equity_curve': equity_curve,
}
__all__ = [
'BacktestTradeRecord',
'StrategyFilter',
'filter_trades',
'run_flat_stake_backtest',
]

188
platform/backend/app/analytics/daily_card_generator.py Normal file
View File

@@ -0,0 +1,188 @@
"""每日智能注單生成器（Daily Smart Card）。"""
from __future__ import annotations
from typing import Any
def _safe_float(value: Any, default: float = 0.0) -> float:
try:
return float(value)
except (TypeError, ValueError):
return default
def _safe_int(value: Any, default: int = 0) -> int:
try:
return int(value)
except (TypeError, ValueError):
return default
def _ev_percent(true_prob: float, decimal_odds: float) -> float:
if decimal_odds <= 1:
return 0.0
implied = 1.0 / decimal_odds
# EV = P*(odds-1) - (1-P)*1
return ((true_prob * (decimal_odds - 1.0)) - (1.0 - true_prob)) * 100
def _guess_stage(match_index: int) -> str:
return '小組賽' if match_index <= 48 else '淘汰賽'
def generate_daily_card(target_date: str, matches: list[dict[str, Any]]) -> dict[str, Any]:
"""
依賽事快照回傳 4 大區塊策略建議（安全單關、搏冷、高勝率串關、同場串關）。
回傳的格式會被前端 /daily-card 與手機版報表一致消化。
"""
safe_singles: list[dict[str, Any]] = []
high_risk_singles: list[dict[str, Any]] = []
safe_parlays: list[dict[str, Any]] = []
sgp_lotteries: list[dict[str, Any]] = []
total_unit = 0.0
for idx, match in enumerate(matches):
match_id = str(match.get('match_id', f'fallback-{idx+1}'))
home_team = str(match.get('home_team', '主隊'))
away_team = str(match.get('away_team', '客隊'))
odds_home = _safe_float(match.get('odds_home'), default=0)
odds_away = _safe_float(match.get('odds_away'), default=0)
# 用 xG 或開盤機率估算真實機率，若無資料則回退到 0.5
home_xg = _safe_float(match.get('home_xg'), default=1.0)
away_xg = _safe_float(match.get('away_xg'), default=1.0)
xg_sum = max(home_xg + away_xg, 0.01)
true_home_prob = home_xg / xg_sum
stage = _guess_stage(idx + 1)
# 安全單關：偏向高勝率市場
if odds_home > 1 and true_home_prob > 0.55:
ev = _ev_percent(true_home_prob, odds_home)
if ev > 3:
safe_unit = 1.8
total_unit += safe_unit
safe_singles.append(
{
'match_id': match_id,
'match_label': f'{home_team} vs {away_team}',
'market_type': '亞洲讓球',
'selection': f'{home_team} -0.25',
'target_odds': round(odds_home, 2),
'win_prob': round(true_home_prob * 100, 2),
'ev_percent': round(ev, 2),
'stake_units': round(safe_unit, 2),
'recommendation': 'SAFE_SINGLE',
'rationale': '高勝率 + 正EV，適合作為核心穩健下注。',
},
)
# 高風險搏冷：低勝率但盤口偏高且 EV 過濾
away_true = 1.0 - true_home_prob
if odds_away > 1 and away_true < 0.35:
ev = _ev_percent(away_true, odds_away)
if ev > 8:
high_risk_unit = 0.35
total_unit += high_risk_unit
high_risk_singles.append(
{
'match_id': match_id,
'match_label': f'{home_team} vs {away_team}',
'market_type': '大小球',
'selection': f'{away_team} 不敗',
'target_odds': round(odds_away, 2),
'win_prob': round(away_true * 100, 2),
'ev_percent': round(ev, 2),
'stake_units': round(high_risk_unit, 2),
'recommendation': 'HIGH_RISK_SINGLE',
'rationale': '冷門高賠率，只有在高勝率組合中保留小倉位。',
},
)
# 2 串 1 串關：選取高勝率的兩個 SAFE 單關，若連乘機率符合條件
if len(safe_singles) >= 2:
legs = safe_singles[:2]
combined_odds = 1.0
combined_prob = 1.0
for leg in legs:
combined_odds *= leg['target_odds']
combined_prob *= leg['win_prob'] / 100
if combined_prob >= 0.28: # 高勝率門檻（保守）
ev = _ev_percent(combined_prob, combined_odds)
if ev > 2:
stake_units = 1.0
total_unit += stake_units
safe_parlays.append(
{
'match_id': 'PARLAY-SAFE',
'match_label': ' + '.join(item['match_label'] for item in legs),
'market_type': '跨場串關',
'selection': '2串1 安全組合',
'legs': [
{
'match_id': item['match_id'],
'selection': item['selection'],
'odds': item['target_odds'],
}
for item in legs
],
'target_odds': round(combined_odds, 2),
'win_prob': round(combined_prob * 100, 2),
'ev_percent': round(ev, 2),
'stake_units': round(stake_units, 2),
'recommendation': 'SAFE_PARLAY',
'rationale': '同風險組合加總，目標追求高穩健率 + 控制回撤。',
'match_stage': _guess_stage(1),
},
)
# 同場 SGP：取出 1 個安全 + 1 個搏冷，形成關聯爆擊模板
if safe_singles and high_risk_singles:
s = safe_singles[0]
h = high_risk_singles[0]
combo_odds = s['target_odds'] * h['target_odds']
combo_prob = (s['win_prob'] / 100) * (h['win_prob'] / 100)
if combo_prob > 0:
ev = _ev_percent(combo_prob, combo_odds)
sgp_lotteries.append(
{
'match_id': s['match_id'],
'match_label': f"{s['match_label']}【同場】",
'market_type': 'SGP',
'selection': f"{s['selection']} + {h['selection']}",
'target_odds': round(combo_odds, 2),
'win_prob': round(combo_prob * 100, 2),
'ev_percent': round(ev, 2),
'stake_units': 0.5,
'recommendation': 'SGP_LOTTERY',
'rationale': '同場串關需監控相關性，避免同向風險重疊。',
'legs': [
{'match_id': s['match_id'], 'selection': s['selection'], 'odds': s['target_odds']},
{'match_id': h['match_id'], 'selection': h['selection'], 'odds': h['target_odds']},
],
'match_stage': _guess_stage(1),
},
)
return {
'date': target_date,
'total_daily_unit_recommendation': round(total_unit, 2),
'summary': (
'系統以當日賽程、赔率變動、xG 進攻強度與場次權重回填，'
'優先輸出高穩定性單關與可控風險的串關建議。'
),
'safe_singles': safe_singles,
'high_risk_singles': high_risk_singles,
'safe_parlays': safe_parlays,
'sgp_lotteries': sgp_lotteries,
'matched_matches': len(matches),
'stage_distribution': {
'小組賽': min(len(matches), 48),
'淘汰賽': max(0, len(matches) - 48),
},
}

112
platform/backend/app/analytics/engine.py Normal file
View File

@@ -0,0 +1,112 @@
"""量化投注引擎（EV、泊松預測、海拔修正）。"""
from __future__ import annotations
import math
import numpy as np
import pandas as pd
from scipy.stats import poisson
def calculate_value_bet(true_prob: float, decimal_odds: float, *, stake: float = 1.0) -> tuple[float, bool]:
"""計算期望值（EV）並判斷是否屬於 Value Bet。
EV 計算：EV = (勝率 * 利潤) - (敗率 * 本金)
其中利潤 = decimal_odds - 1。
Returns
-------
ev_pct: float
以本金為基底的 EV 百分比（EV / stake）。
is_value_bet: bool
當 EV > 0.03（3%）回傳 True。
"""
prob = float(true_prob)
odds = float(decimal_odds)
if not 0 <= prob <= 1 or odds <= 1 or stake <= 0:
return 0.0, False
profit = odds - 1
ev = prob * profit - (1 - prob) * stake
ev_pct = ev / stake
return round(ev_pct, 6), ev_pct > 0.03
class PoissonPredictor:
"""球員進球分佈預測器（2x2 進球建模）。"""
def __init__(
self,
home_attack: float,
home_defense: float,
away_attack: float,
away_defense: float,
league_avg_goals: float,
) -> None:
self.home_attack = float(home_attack)
self.home_defense = float(home_defense)
self.away_attack = float(away_attack)
self.away_defense = float(away_defense)
self.league_avg_goals = float(league_avg_goals)
# 以攻守乘積估算 λ，並限制在合理範圍避免極端值發散。
home_lambda = league_avg_goals * (self.home_attack / max(self.away_defense, 0.01))
away_lambda = league_avg_goals * (self.away_attack / max(self.home_defense, 0.01))
self.home_lambda = float(np.clip(home_lambda, 0.02, 6.5))
self.away_lambda = float(np.clip(away_lambda, 0.02, 6.5))
def predict_exact_score(self, home_goals: int, away_goals: int) -> float:
"""回傳指定波膽（home_goals, away_goals）發生機率。"""
p_home = poisson.pmf(home_goals, self.home_lambda)
p_away = poisson.pmf(away_goals, self.away_lambda)
return float(p_home * p_away)
def predict_over_under_prob(self, line: float = 2.5, max_goals: int = 10) -> tuple[float, float]:
"""回傳（under, over）機率。"""
goals = pd.MultiIndex.from_product(
[range(max_goals + 1), range(max_goals + 1)],
names=['home', 'away'],
).to_frame(index=False)
def joint_prob(r: pd.Series) -> float:
return float(poisson.pmf(r['home'], self.home_lambda) * poisson.pmf(r['away'], self.away_lambda))
probs = goals.apply(joint_prob, axis=1)
total_goals = goals['home'] + goals['away']
under = float(probs[total_goals <= line].sum())
over = float(probs[total_goals > line].sum())
return under, over
def adjust_away_defense_for_altitude(
base_defense_rating: float,
venue_altitude_meters: float,
*,
is_second_half: bool,
penalty_factor: float = 0.35,
) -> float:
"""高海拔下修正客隊防守能力。
當場地海拔高於 1500m 且處於下半場，套用對數懲罰，
代表客隊在氧氣濃度降低下體能下降導致防守效率衰退。
"""
base = float(base_defense_rating)
if venue_altitude_meters <= 1500 or not is_second_half:
return base
# 以 log(1 + altitude/1000) 做平滑遞增函式，避免低海拔時劇烈改變。
altitude_penalty = penalty_factor * math.log1p(venue_altitude_meters / 1000)
return base * (1 - min(max(altitude_penalty, 0), 0.45))
__all__ = [
'calculate_value_bet',
'PoissonPredictor',
'adjust_away_defense_for_altitude',
]

43
platform/backend/app/analytics/environment_model.py Normal file
View File

@@ -0,0 +1,43 @@
"""比賽環境衰減模型（高海拔與高溫）。"""
from __future__ import annotations
import math
def adjust_team_strength_for_environment(
base_strength: float,
venue_altitude: float,
venue_heat_index: float,
is_second_half: bool,
team_acclimatized: bool,
) -> float:
"""調整球隊能力值，反映環境壓力。
- 海拔 > 1500m 且球隊未適應，第二節時套用疲勞衰退。
- 熱指數（Heat Index）越高，衰退越明顯。
"""
if base_strength < 0:
raise ValueError('base_strength 必須大於等於 0')
adjusted = float(base_strength)
if not is_second_half:
return adjusted
altitude_penalty = 0.0
heat_penalty = 0.0
if not team_acclimatized and venue_altitude >= 1500:
# 以對數遞增，1500m 為轉折，3000m 接近上限。
altitude_factor = math.log1p((venue_altitude - 1500.0) / 300.0)
altitude_penalty = 0.025 + 0.045 * min(altitude_factor, 2.8)
# 熱指數高於 30，逐步加入疲勞因子，超過 38 非常明顯。
if venue_heat_index > 30:
heat_excess = min(max(venue_heat_index - 30.0, 0.0), 30.0)
heat_penalty = 0.0012 * heat_excess
total_penalty = altitude_penalty + heat_penalty
adjusted *= max(0.2, 1.0 - total_penalty)
return adjusted

63
platform/backend/app/analytics/ev_calculator.py Normal file
View File

@@ -0,0 +1,63 @@
"""EV（期望值）運算模組。
本模組提供最基本、可復用的賠率價值判斷邏輯：給定真實勝率與小數賠率，計算期望值與是否為優勢投注。
"""
from __future__ import annotations
from typing import Any
def calculate_expected_value(
true_win_prob: float,
decimal_odds: float,
stake: float = 100.0,
suggested_kelly_fraction: float | None = None,
) -> dict[str, Any]:
"""計算期望值（EV）並回傳報價建議。
Parameters
----------
true_win_prob:
模型估計的真實勝率，必須在 0 到 1 之間。
decimal_odds:
小數制賠率，必須大於 1（否則不具可投注意義）。
stake:
本次下注本金；同時也是 EV 百分比的基準。
suggested_kelly_fraction:
由外部凱利公式模組預留的建議資金比例；若未提供則回傳 None。
Returns
-------
dict
{
'ev_value': 實際 EV 金額,
'ev_percentage': EV / stake * 100,
'is_value_bet': 當 EV% 大於 3% 時為 True,
'suggested_kelly_fraction': 傳入值或 None
}
"""
if not 0.0 <= true_win_prob <= 1.0:
raise ValueError('true_win_prob 必須介於 0 到 1 之間')
if decimal_odds <= 1:
raise ValueError('decimal_odds 必須大於 1')
if stake <= 0:
raise ValueError('stake 必須大於 0')
win_prob = float(true_win_prob)
odds = float(decimal_odds)
stake_amount = float(stake)
profit_when_win = odds - 1.0
lose_prob = 1.0 - win_prob
ev = win_prob * profit_when_win * stake_amount - lose_prob * stake_amount
ev_percentage = ev / stake_amount * 100
return {
'ev_value': round(ev, 6),
'ev_percentage': round(ev_percentage, 4),
'is_value_bet': ev_percentage > 3.0,
'suggested_kelly_fraction': suggested_kelly_fraction,
}

163
platform/backend/app/analytics/feature_engineering.py Normal file
View File

@@ -0,0 +1,163 @@
"""進階特徵工程：從資料庫抽取多維比賽特徵。"""
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime
from math import radians, sin, cos, asin, sqrt
from typing import Iterable
from sqlalchemy import and_, desc, select
from sqlalchemy.ext.asyncio import AsyncSession
from ..db.models import Match, Team
@dataclass(frozen=True)
class MatchFeatureVector:
rest_days_advantage: float
travel_distance_km: float
recent_5_xg_diff: float
elo_rating_diff: float
def _haversine_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
"""Haversine 地球大圓距離（公里）。"""
R = 6371.0
dlat = radians(lat2 - lat1)
dlon = radians(lon2 - lon1)
a = sin(dlat / 2) ** 2 + cos(radians(lat1)) * cos(radians(lat2)) * sin(dlon / 2) ** 2
return 2 * R * asin(min(1.0, sqrt(a)))
class MatchFeatureExtractor:
"""抽取並生成賽前特徵。"""
def __init__(
self,
session_factory,
*,
team_locations: dict[str, tuple[float, float]] | None = None,
) -> None:
self.session_factory = session_factory
# 可選：{team_id: (lat, lon)}，若缺資料則 fallback 為 0 距離。
self.team_locations = team_locations or {}
async def _previous_match(self, session: AsyncSession, team_id: str, match_time: datetime) -> Match | None:
stmt = (
select(Match)
.where(
and_(
(Match.home_team_id == team_id) | (Match.away_team_id == team_id),
Match.match_time_utc < match_time,
Match.home_xg.is_not(None),
Match.away_xg.is_not(None),
),
)
.order_by(desc(Match.match_time_utc))
.limit(1)
)
result = await session.execute(stmt)
return result.scalar_one_or_none()
async def _recent_xg_series(self, session: AsyncSession, team_id: str, as_of_match_id: str, count: int = 5) -> list[float]:
stmt = (
select(Match)
.where(
(Match.home_team_id == team_id) | (Match.away_team_id == team_id),
Match.home_xg.is_not(None),
Match.away_xg.is_not(None),
Match.id != as_of_match_id,
)
.order_by(desc(Match.match_time_utc))
.limit(count)
)
result = await session.execute(stmt)
rows = result.scalars().all()
out: list[float] = []
for row in rows:
home_xg = float(row.home_xg or 0.0)
away_xg = float(row.away_xg or 0.0)
out.append(home_xg)
out.append(away_xg)
return out[:count]
async def extract_features(self, match_id: str) -> MatchFeatureVector:
"""產生四個關鍵特徵。
1) rest_days_advantage
2) travel_distance_km
3) recent_5_xg_diff
4) elo_rating_diff
"""
async with self.session_factory() as session: # type: ignore[assignment]
current_match = await session.get(Match, match_id)
if current_match is None:
raise ValueError(f'找不到 match_id={match_id}')
home_team = await session.get(Team, current_match.home_team_id)
away_team = await session.get(Team, current_match.away_team_id)
if home_team is None or away_team is None:
raise ValueError('比賽球隊資料不完整')
home_prev = await self._previous_match(session, home_team.id, current_match.match_time_utc)
away_prev = await self._previous_match(session, away_team.id, current_match.match_time_utc)
rest_home = (
(current_match.match_time_utc - home_prev.match_time_utc).days
if home_prev is not None
else 0
)
rest_away = (
(current_match.match_time_utc - away_prev.match_time_utc).days
if away_prev is not None
else 0
)
travel_distance = self._distance_between_teams(home_team.id, away_team.id)
home_xg = await self._recent_xg_series(session, home_team.id, current_match.id)
away_xg = await self._recent_xg_series(session, away_team.id, current_match.id)
recent_diff = sum(home_xg[:5]) / max(len(home_xg[:5]) or 1, 1) - sum(away_xg[:5]) / max(
len(away_xg[:5]) or 1,
1,
)
home_elo = float(home_team.current_elo_rating or 1500)
away_elo = float(away_team.current_elo_rating or 1500)
return MatchFeatureVector(
rest_days_advantage=float(rest_home - rest_away),
travel_distance_km=float(travel_distance),
recent_5_xg_diff=float(recent_diff),
elo_rating_diff=float(home_elo - away_elo),
)
def _distance_between_teams(self, home_team_id: str, away_team_id: str) -> float:
home_loc = self.team_locations.get(home_team_id)
away_loc = self.team_locations.get(away_team_id)
if home_loc is None or away_loc is None:
return 0.0
return float(_haversine_km(home_loc[0], home_loc[1], away_loc[0], away_loc[1]))
@staticmethod
def to_model_payload(features: MatchFeatureVector, columns: Iterable[str] | None = None) -> dict:
"""輸出可直接餵進 XGBoost 的特徵字典。"""
payload = {
'rest_days_advantage': features.rest_days_advantage,
'travel_distance_km': features.travel_distance_km,
'recent_5_xg_diff': features.recent_5_xg_diff,
'elo_rating_diff': features.elo_rating_diff,
}
if columns is None:
return payload
cols = list(columns)
return {c: float(payload[c]) for c in cols if c in payload}

39
platform/backend/app/analytics/hedging_calculator.py Normal file
View File

@@ -0,0 +1,39 @@
"""串關動態對沖（Dynamic Hedging）計算器。"""
from __future__ import annotations
def calculate_hedge_amount(
original_stake: float,
parlay_total_odds: float,
final_leg_hedge_odds: float,
) -> dict[str, float]:
"""
在 1 場或 2/3 場連贏快到最終局，計算對沖下注金額。
將原始串關保本化：
目標是「串關全過的淨利」與「對沖走向中的淨利」在最後同值。
設原始串關到位後保底淨利 = S * (O_parlay - 1)
對沖選項淨利 = H * (O_hedge - 1)
求 H * (O_hedge - 1) = S * (O_parlay - 1) - H
=> H = (S * (O_parlay - 1)) / O_hedge
"""
if original_stake <= 0:
raise ValueError('original_stake 必須大於 0')
if parlay_total_odds <= 1:
raise ValueError('parlay_total_odds 必須大於 1')
if final_leg_hedge_odds <= 1:
raise ValueError('final_leg_hedge_odds 必須大於 1')
expected_parlay_net = original_stake * (parlay_total_odds - 1)
hedge_stake = expected_parlay_net / final_leg_hedge_odds
profit_after_hedge = hedge_stake * (final_leg_hedge_odds - 1)
return {
'hedge_stake': round(hedge_stake, 4),
'locked_profit': round(profit_after_hedge, 4),
'parlay_net_after_hedge_if_win': round(expected_parlay_net - hedge_stake, 4),
'hedge_net_if_win': round(profit_after_hedge, 4),
}

64
platform/backend/app/analytics/kelly.py Normal file
View File

@@ -0,0 +1,64 @@
"""凱利準則（Kelly Criterion）工具。"""
from __future__ import annotations
from dataclasses import dataclass
@dataclass(frozen=True)
class KellyResult:
"""凱利投注建議結果。"""
decimal_odds: float
win_probability: float
raw_kelly_fraction: float
fractional_kelly_factor: float
risk_tolerance_factor: float
final_fraction: float
stake_fraction: float
def calculate_kelly_fraction(
decimal_odds: float,
true_prob: float,
*,
bankroll: float,
fractional_kelly_factor: float = 1.0,
risk_tolerance_factor: float = 1.0,
) -> KellyResult:
"""依凱利準則估算下注比例與建議金額。
凱利公式：
f* = (b * p - q) / b
其中 b = odds - 1，p 為勝率，q = 1 - p。
"""
if decimal_odds <= 1:
raise ValueError('decimal_odds 必須大於 1')
if bankroll <= 0:
raise ValueError('bankroll 必須大於 0')
if not 0 <= true_prob <= 1:
raise ValueError('true_prob 需介於 0 到 1')
if not 0 <= fractional_kelly_factor <= 5:
raise ValueError('fractional_kelly_factor 須介於 0 到 5')
if not 0 <= risk_tolerance_factor <= 2:
raise ValueError('risk_tolerance_factor 須介於 0 到 2')
b = decimal_odds - 1
raw_kelly = (b * true_prob - (1 - true_prob)) / b
final_fraction = raw_kelly * fractional_kelly_factor * risk_tolerance_factor
# 保守處理：避免負值與超過總資金比例（100%）的極端輸出。
final_fraction = max(0.0, min(final_fraction, 1.0))
return KellyResult(
decimal_odds=decimal_odds,
win_probability=true_prob,
raw_kelly_fraction=raw_kelly,
fractional_kelly_factor=fractional_kelly_factor,
risk_tolerance_factor=risk_tolerance_factor,
final_fraction=final_fraction,
stake_fraction=final_fraction,
)
__all__ = ['KellyResult', 'calculate_kelly_fraction']

435
platform/backend/app/analytics/ml_ensemble.py Normal file
View File

@@ -0,0 +1,435 @@
"""機器學習賽果預測引擎（Ensemble）。"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Mapping
from uuid import uuid4
import numpy as np
import pandas as pd
try:
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.model_selection import train_test_split
except Exception: # pragma: no cover - 缺少 scikit-learn 時的 fallback
GradientBoostingClassifier = None
train_test_split = None
FEATURE_COLUMNS = ('rest_days_advantage', 'travel_distance_km', 'recent_5_xg_diff')
OUTCOMES = ('home', 'draw', 'away')
def _sigmoid(value: float) -> float:
return 1.0 / (1.0 + np.exp(-value))
def _softmax(values: np.ndarray) -> np.ndarray:
shifted = values - np.max(values)
exp_values = np.exp(shifted)
return exp_values / exp_values.sum()
@dataclass(frozen=True)
class EnsembleModelArtifact:
"""已訓練的 ML 模組與中繼資料。"""
model: Any
feature_columns: tuple[str, ...]
model_id: str
training_size: int
is_fallback: bool
training_accuracy: float | None = None
class _FallbackMatchModel:
"""缺少 ML 套件時的保底模型（規則式）。"""
feature_columns = FEATURE_COLUMNS
def predict_proba(self, row_df: pd.DataFrame) -> np.ndarray:
if row_df.empty:
return np.zeros((0, 3))
x = row_df[self.feature_columns].to_numpy(float)
raw_scores = []
for rest_days_advantage, travel_distance_km, recent_5_xg_diff in x:
home_score = 0.6 + rest_days_advantage * 0.022 + recent_5_xg_diff * 0.34 - travel_distance_km * 0.0012
draw_score = 0.30 - abs(rest_days_advantage) * 0.015 - abs(recent_5_xg_diff) * 0.22
away_score = 0.1 - rest_days_advantage * 0.022 - recent_5_xg_diff * 0.34 + travel_distance_km * 0.0012
scores = np.array(
[
_sigmoid(home_score),
_sigmoid(draw_score) * 0.9,
_sigmoid(away_score),
],
dtype=float,
)
raw_scores.append(_softmax(scores))
return np.vstack(raw_scores)
def _as_float(value: Any, default: float = 0.0) -> float:
try:
return float(value)
except (TypeError, ValueError):
return default
def normalize_feature_payload(payload: Mapping[str, Any]) -> dict[str, float]:
"""從前端或資料庫欄位，萃取核心三大特徵。"""
home_rest = _as_float(payload.get('home_rest_days'))
away_rest = _as_float(payload.get('away_rest_days'))
home_travel = _as_float(payload.get('home_travel_distance_km'))
away_travel = _as_float(payload.get('away_travel_distance_km'))
recent_home = _as_float(payload.get('recent_5_xg_home'))
recent_away = _as_float(payload.get('recent_5_xg_away'))
return {
'home_rest_days': home_rest,
'away_rest_days': away_rest,
'home_travel_distance_km': home_travel,
'away_travel_distance_km': away_travel,
'recent_5_xg_home': recent_home,
'recent_5_xg_away': recent_away,
'rest_days_advantage': home_rest - away_rest,
'travel_distance_km': home_travel - away_travel,
'recent_5_xg_diff': recent_home - recent_away,
}
def _validation_frame(rows: list[Mapping[str, Any]]) -> pd.DataFrame:
if len(rows) < 5:
raise ValueError('訓練樣本少於 5 筆，無法完成穩定訓練')
frame = pd.DataFrame(rows)
required_fields = set(FEATURE_COLUMNS) | {'match_result'}
missing = required_fields - set(frame.columns)
if missing:
raise ValueError(f'訓練資料缺欄位：{sorted(missing)}')
frame = frame.copy()
frame[list(FEATURE_COLUMNS)] = frame[list(FEATURE_COLUMNS)].astype(float).fillna(0.0)
frame['match_result'] = frame['match_result'].str.lower().str.strip()
unknown = set(frame['match_result']) - set(OUTCOMES)
if unknown:
raise ValueError(f'未知賽果標籤：{sorted(unknown)}，僅支援 {OUTCOMES}')
return frame
def build_default_ml_training_rows() -> list[dict[str, float | str]]:
"""建立保底訓練樣本（當環境無法即時取得外部訓練資料時）。"""
return [
{
'home_rest_days': 4,
'away_rest_days': 3,
'home_travel_distance_km': 520,
'away_travel_distance_km': 1100,
'recent_5_xg_home': 1.8,
'recent_5_xg_away': 1.0,
'rest_days_advantage': 1,
'travel_distance_km': -580,
'recent_5_xg_diff': 0.8,
'match_result': 'home',
},
{
'home_rest_days': 2,
'away_rest_days': 5,
'home_travel_distance_km': 220,
'away_travel_distance_km': 780,
'recent_5_xg_home': 1.1,
'recent_5_xg_away': 1.7,
'rest_days_advantage': -3,
'travel_distance_km': -560,
'recent_5_xg_diff': -0.6,
'match_result': 'away',
},
{
'home_rest_days': 6,
'away_rest_days': 4,
'home_travel_distance_km': 120,
'away_travel_distance_km': 960,
'recent_5_xg_home': 2.3,
'recent_5_xg_away': 1.8,
'rest_days_advantage': 2,
'travel_distance_km': -840,
'recent_5_xg_diff': 0.5,
'match_result': 'home',
},
{
'home_rest_days': 3,
'away_rest_days': 3,
'home_travel_distance_km': 900,
'away_travel_distance_km': 900,
'recent_5_xg_home': 1.2,
'recent_5_xg_away': 1.3,
'rest_days_advantage': 0,
'travel_distance_km': 0,
'recent_5_xg_diff': -0.1,
'match_result': 'draw',
},
{
'home_rest_days': 8,
'away_rest_days': 2,
'home_travel_distance_km': 350,
'away_travel_distance_km': 700,
'recent_5_xg_home': 2.0,
'recent_5_xg_away': 1.2,
'rest_days_advantage': 6,
'travel_distance_km': -350,
'recent_5_xg_diff': 0.8,
'match_result': 'home',
},
{
'home_rest_days': 1,
'away_rest_days': 2,
'home_travel_distance_km': 1600,
'away_travel_distance_km': 2500,
'recent_5_xg_home': 1.4,
'recent_5_xg_away': 2.1,
'rest_days_advantage': -1,
'travel_distance_km': -900,
'recent_5_xg_diff': -0.7,
'match_result': 'away',
},
{
'home_rest_days': 5,
'away_rest_days': 5,
'home_travel_distance_km': 700,
'away_travel_distance_km': 700,
'recent_5_xg_home': 1.9,
'recent_5_xg_away': 1.9,
'rest_days_advantage': 0,
'travel_distance_km': 0,
'recent_5_xg_diff': 0.0,
'match_result': 'draw',
},
{
'home_rest_days': 9,
'away_rest_days': 3,
'home_travel_distance_km': 400,
'away_travel_distance_km': 300,
'recent_5_xg_home': 2.4,
'recent_5_xg_away': 1.1,
'rest_days_advantage': 6,
'travel_distance_km': 100,
'recent_5_xg_diff': 1.3,
'match_result': 'home',
},
{
'home_rest_days': 2,
'away_rest_days': 7,
'home_travel_distance_km': 1800,
'away_travel_distance_km': 250,
'recent_5_xg_home': 1.0,
'recent_5_xg_away': 1.5,
'rest_days_advantage': -5,
'travel_distance_km': 1550,
'recent_5_xg_diff': -0.5,
'match_result': 'away',
},
{
'home_rest_days': 4,
'away_rest_days': 4,
'home_travel_distance_km': 500,
'away_travel_distance_km': 500,
'recent_5_xg_home': 1.6,
'recent_5_xg_away': 1.4,
'rest_days_advantage': 0,
'travel_distance_km': 0,
'recent_5_xg_diff': 0.2,
'match_result': 'home',
},
{
'home_rest_days': 6,
'away_rest_days': 1,
'home_travel_distance_km': 300,
'away_travel_distance_km': 1200,
'recent_5_xg_home': 2.8,
'recent_5_xg_away': 0.8,
'rest_days_advantage': 5,
'travel_distance_km': -900,
'recent_5_xg_diff': 2.0,
'match_result': 'home',
},
{
'home_rest_days': 2,
'away_rest_days': 6,
'home_travel_distance_km': 1000,
'away_travel_distance_km': 200,
'recent_5_xg_home': 1.0,
'recent_5_xg_away': 2.6,
'rest_days_advantage': -4,
'travel_distance_km': 800,
'recent_5_xg_diff': -1.6,
'match_result': 'away',
},
{
'home_rest_days': 7,
'away_rest_days': 7,
'home_travel_distance_km': 650,
'away_travel_distance_km': 650,
'recent_5_xg_home': 1.8,
'recent_5_xg_away': 1.8,
'rest_days_advantage': 0,
'travel_distance_km': 0,
'recent_5_xg_diff': 0.0,
'match_result': 'draw',
},
{
'home_rest_days': 3,
'away_rest_days': 1,
'home_travel_distance_km': 260,
'away_travel_distance_km': 900,
'recent_5_xg_home': 2.1,
'recent_5_xg_away': 1.6,
'rest_days_advantage': 2,
'travel_distance_km': -640,
'recent_5_xg_diff': 0.5,
'match_result': 'home',
},
{
'home_rest_days': 0,
'away_rest_days': 5,
'home_travel_distance_km': 1500,
'away_travel_distance_km': 150,
'recent_5_xg_home': 1.2,
'recent_5_xg_away': 2.0,
'rest_days_advantage': -5,
'travel_distance_km': 1350,
'recent_5_xg_diff': -0.8,
'match_result': 'away',
},
{
'home_rest_days': 5,
'away_rest_days': 2,
'home_travel_distance_km': 300,
'away_travel_distance_km': 300,
'recent_5_xg_home': 2.2,
'recent_5_xg_away': 1.1,
'rest_days_advantage': 3,
'travel_distance_km': 0,
'recent_5_xg_diff': 1.1,
'match_result': 'home',
},
{
'home_rest_days': 4,
'away_rest_days': 8,
'home_travel_distance_km': 450,
'away_travel_distance_km': 980,
'recent_5_xg_home': 1.5,
'recent_5_xg_away': 2.4,
'rest_days_advantage': -4,
'travel_distance_km': -530,
'recent_5_xg_diff': -0.9,
'match_result': 'away',
},
]
def train_match_outcome_ensemble(
training_rows: list[Mapping[str, Any]],
*,
model_id: str | None = None,
) -> EnsembleModelArtifact:
"""訓練 1X2 賽果 Ensemble（無法使用 sklearn 時自動回退規則模型）。"""
normalized = [_normalize_training_row(row) for row in training_rows]
frame = _validation_frame(normalized)
x = frame[list(FEATURE_COLUMNS)]
y = frame['match_result'].map({'home': 0, 'draw': 1, 'away': 2})
if len(frame) < 24 or GradientBoostingClassifier is None or train_test_split is None:
return EnsembleModelArtifact(
model=_FallbackMatchModel(),
feature_columns=FEATURE_COLUMNS,
model_id=model_id or uuid4().hex,
training_size=len(frame),
is_fallback=True,
training_accuracy=None,
)
x_train, x_val, y_train, y_val = train_test_split(
x,
y,
test_size=min(0.3, max(0.15, 1 - (30 / len(frame)))),
random_state=17,
stratify=y,
)
model = GradientBoostingClassifier(
random_state=17,
n_estimators=220,
max_depth=3,
learning_rate=0.06,
)
model.fit(x_train, y_train)
accuracy = float(model.score(x_val, y_val)) if len(set(y_val)) > 1 else None
return EnsembleModelArtifact(
model=model,
feature_columns=FEATURE_COLUMNS,
model_id=model_id or uuid4().hex,
training_size=len(frame),
is_fallback=False,
training_accuracy=accuracy,
)
def _normalize_training_row(row: Mapping[str, Any]) -> dict[str, float | str]:
normalized = normalize_feature_payload(row)
if 'match_result' not in row:
raise ValueError('訓練資料缺少 match_result')
normalized['match_result'] = str(row['match_result']).strip().lower()
return normalized
def build_default_ensemble_artifact() -> EnsembleModelArtifact:
"""建立系統預設模型（含 fallback）。"""
return train_match_outcome_ensemble(build_default_ml_training_rows(), model_id='default')
def model_predict_probabilities(
artifact: EnsembleModelArtifact,
features: Mapping[str, Any],
) -> dict[str, float]:
"""回傳 home/draw/away 的機率。"""
normalized = normalize_feature_payload(features)
feature_frame = pd.DataFrame([normalized], columns=artifact.feature_columns)
probs = artifact.model.predict_proba(feature_frame)[0]
return {
'home': float(probs[0]),
'draw': float(probs[1]),
'away': float(probs[2]),
}
def calculate_model_edges(
predicted: dict[str, float],
implied: dict[str, float],
) -> dict[str, dict[str, float | bool]]:
"""比較模型機率與莊家隱含機率，標示 Strong Buy。"""
edges: dict[str, dict[str, float | bool]] = {}
for key in OUTCOMES:
p = float(predicted.get(key, 0))
i = float(implied.get(key, 0))
edge = p - i
edges[key] = {
'model_prob': round(p, 6),
'implied_prob': round(i, 6),
'edge': round(edge, 6),
'strong_buy': edge >= 0.04,
}
return edges

99
platform/backend/app/analytics/ml_inference.py Normal file
View File

@@ -0,0 +1,99 @@
"""XGBoost 推論 API 套件。"""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import Any
import numpy as np
from xgboost import Booster, DMatrix
@dataclass(frozen=True)
class XGBoostPrediction:
home_win: float
draw: float
away_win: float
def _safe_probability(x: float) -> float:
return float(max(0.0, min(1.0, x)))
class XGBoostPredictor:
"""XGBoost 預測器：輸入特徵 => 輸出 1x2 機率。"""
def __init__(
self,
model_path: str | None = None,
*,
feature_columns: list[str] | None = None,
) -> None:
self.feature_columns = feature_columns or []
self.model_path = model_path
self.model = self._load_model(model_path) if model_path else None
def _load_model(self, model_path: str | None) -> Booster | None:
if not model_path:
return None
path = Path(model_path)
if not path.exists():
return None
model = Booster()
model.load_model(str(path))
return model
def predict_match_outcome(self, features: dict[str, float]) -> dict[str, float]:
"""輸出主勝/平/客勝機率。"""
if self.model is None:
# fallback: 均分
return {'home': 1 / 3, 'draw': 1 / 3, 'away': 1 / 3}
ordered_values = [float(features.get(col, 0.0)) for col in self.feature_columns]
dmatrix = DMatrix(np.array([ordered_values]), feature_names=self.feature_columns)
probs = self.model.predict(dmatrix)
if probs.ndim == 1:
probs = probs.reshape(1, -1)
arr = probs[0]
if arr.size < 3:
raise ValueError('模型輸出維度不足 3')
raw = np.array(arr[:3], dtype=float)
raw = np.maximum(raw, 0.0)
s = raw.sum()
if s <= 0:
raise ValueError('模型輸出總和異常為 0')
norm = raw / s
return {'home': _safe_probability(norm[0]), 'draw': _safe_probability(norm[1]), 'away': _safe_probability(norm[2])}
def find_model_edge(
self,
ml_probs: dict[str, float],
bookmaker_implied_probs: dict[str, float],
) -> list[dict[str, Any]]:
"""回傳模型超越莊家 4% 以上的投注選項。"""
mapping = [('home', 'home'), ('draw', 'draw'), ('away', 'away')]
outputs: list[dict[str, Any]] = []
for model_key, book_key in mapping:
ml_v = float(ml_probs.get(model_key, 0.0))
book_v = float(bookmaker_implied_probs.get(book_key, 0.0))
edge = ml_v - book_v
if edge >= 0.04:
outputs.append(
{
'selection': model_key,
'ml_prob': round(ml_v, 6),
'bookmaker_implied_prob': round(book_v, 6),
'edge': round(edge, 6),
'label': 'Strong Buy',
},
)
return outputs

163
platform/backend/app/analytics/player_props.py Normal file
View File

@@ -0,0 +1,163 @@
"""球員道具盤（Player Props）量化引擎。"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Literal
import numpy as np
PropMetric = Literal['shots', 'shots_on_target', 'passes']
@dataclass(frozen=True)
class PlayerPropsProfile:
"""球員與對位環境的道具盤參考參數。"""
player_id: str
metric: PropMetric
baseline_mean: float
match_minutes: int = 90
team_attack_factor: float = 1.0
opponent_defence_factor: float = 1.0
weather_fatigue_factor: float = 1.0
@dataclass(frozen=True)
class PlayerPropsSimulationResult:
"""單個道具盤的模擬輸出。"""
metric: PropMetric
line: float
over_probability: float
under_probability: float
expected_count: float
p5: float
p50: float
p95: float
simulation_runs: int
def to_dict(self) -> dict[str, float | int | str]:
return {
'metric': self.metric,
'line': self.line,
'over_probability': self.over_probability,
'under_probability': self.under_probability,
'expected_count': self.expected_count,
'p5': self.p5,
'p50': self.p50,
'p95': self.p95,
'simulation_runs': self.simulation_runs,
}
def _apply_context_multiplier(profile: PlayerPropsProfile) -> float:
"""依據球員對位環境組合出單場事件期望值修正係數。"""
multipliers = [
max(0.1, profile.team_attack_factor),
1 / max(0.5, profile.opponent_defence_factor),
max(0.6, profile.weather_fatigue_factor),
]
return float(np.prod(multipliers))
def _metric_seed_variance(profile: PlayerPropsProfile) -> float:
"""使用不同維度的離散程度（sigma）以保留球員特徵差異。"""
if profile.metric == 'passes':
return 0.45
if profile.metric == 'shots_on_target':
return 0.22
return 0.30
def simulate_player_prop_probability(
profile: PlayerPropsProfile,
*,
line: float,
simulations: int = 10000,
rng: np.random.Generator | None = None,
) -> PlayerPropsSimulationResult:
"""用蒙地卡羅法計算球員道具盤超過盤口線的機率。"""
if line <= 0:
raise ValueError('line 必須為正數')
if simulations <= 100:
raise ValueError('simulations 最少需要 100 次')
generator = rng or np.random.default_rng()
minute_ratio = profile.match_minutes / 90
base = profile.baseline_mean * minute_ratio
adjusted_mean = max(0.05, base * _apply_context_multiplier(profile))
# 以 Gamma-Poisson 混合近似捕捉波動，避免單純 Poisson 太過平滑。
gamma_shape = max(0.5, 1.0 / (_metric_seed_variance(profile) ** 2))
gamma_scale = adjusted_mean / gamma_shape
intensity = generator.gamma(gamma_shape, gamma_scale, size=simulations)
counts = generator.poisson(intensity).astype(float)
over_count = int(np.sum(counts > line))
over_probability = over_count / simulations
under_probability = 1 - over_probability
expected_count = float(np.mean(counts))
p5, p50, p95 = [float(np.percentile(counts, q)) for q in (5, 50, 95)]
return PlayerPropsSimulationResult(
metric=profile.metric,
line=line,
over_probability=round(over_probability, 6),
under_probability=round(under_probability, 6),
expected_count=round(expected_count, 3),
p5=p5,
p50=p50,
p95=p95,
simulation_runs=simulations,
)
def evaluate_top_edge(
profile: PlayerPropsProfile,
bookmaker_over_odds: float,
*,
line: float,
simulations: int = 10000,
stake: float = 1.0,
) -> dict[str, Any]:
"""回傳道具盤 EV 與建議邊際，供前端高邊際卡片使用。"""
result = simulate_player_prop_probability(profile, line=line, simulations=simulations)
if bookmaker_over_odds <= 1:
raise ValueError('bookmaker_over_odds 必須大於 1')
# EV 計算以 "賭 over" 為例。
win_profit = (bookmaker_over_odds - 1) * stake
loss = stake
ev = result.over_probability * win_profit - (1 - result.over_probability) * loss
edge = ev / stake
top_edge = edge > 0.08
return {
**result.to_dict(),
'edge': round(edge, 6),
'top_edge': top_edge,
'bookmaker_over_odds': bookmaker_over_odds,
'implied_prob': round(1 / bookmaker_over_odds, 6),
'recommended_stake_hint': round(max(0.0, edge * stake * 0.4), 2),
}
__all__ = [
'PropMetric',
'PlayerPropsProfile',
'PlayerPropsSimulationResult',
'evaluate_top_edge',
'simulate_player_prop_probability',
]

79
platform/backend/app/analytics/player_props_sim.py Normal file
View File

@@ -0,0 +1,79 @@
"""球員道具盤（Props）蒙地卡羅模擬模組。"""
from __future__ import annotations
from dataclasses import dataclass
import numpy as np
@dataclass(frozen=True)
class PlayerPropsDistribution:
shots: np.ndarray
shots_on_target: np.ndarray
passes: np.ndarray
def simulate_player_stats(
player_metrics: dict,
opponent_defense_metrics: dict,
iterations: int = 10_000,
) -> PlayerPropsDistribution:
"""快速模擬球員事件次數分佈。"""
if iterations <= 0:
raise ValueError('iterations 必須大於 0')
avg_touches = float(player_metrics.get('avg_touches', 45) or 0.0)
base_shot_rate = float(player_metrics.get('shots_per_touch', 0.08) or 0.0)
base_target_rate = float(player_metrics.get('shot_on_target_rate', 0.35) or 0.0)
base_pass_rate = float(player_metrics.get('passes_per_touch', 0.65) or 0.0)
opp_pressure = float(opponent_defense_metrics.get('pressing_index', 1.0) or 1.0)
opp_tackling = float(opponent_defense_metrics.get('marking_index', 1.0) or 1.0)
adj_touches = max(1.0, avg_touches * max(0.6, 1.0 / max(0.5, opp_pressure)))
shot_lambda = adj_touches * base_shot_rate
pass_lambda = adj_touches * base_pass_rate
rng = np.random.default_rng()
shots = rng.poisson(lam=shot_lambda, size=iterations)
passes = rng.poisson(lam=pass_lambda, size=iterations)
# 對方壓迫會降低射正率
effective_target_rate = max(0.02, base_target_rate / max(opp_tackling, 0.3))
shots_on_target = rng.binomial(shots, p=min(effective_target_rate, 0.99), size=iterations)
return PlayerPropsDistribution(shots=shots.astype(int), shots_on_target=shots_on_target.astype(int), passes=passes.astype(int))
def evaluate_prop_bet(
simulated_distribution: PlayerPropsDistribution,
line: float,
odds: float,
) -> dict[str, float | bool]:
"""從 10,000 次模擬結果計算超過盤口機率與 EV。"""
if odds <= 1:
raise ValueError('odds 必須大於 1')
if line < 0:
raise ValueError('line 必須大於等於 0')
shots = simulated_distribution.shots
if shots.size == 0:
raise ValueError('distribution 為空')
probability_over = float((shots > line).mean())
from .ev_calculator import calculate_expected_value
ev = calculate_expected_value(probability_over, odds)
return {
'metric': 'shots',
'line': line,
'over_probability': round(probability_over, 6),
'under_probability': round(1.0 - probability_over, 6),
'implied_ev': ev['ev_value'],
'ev_percentage': ev['ev_percentage'],
'is_value_bet': bool(ev['is_value_bet']),
}

113
platform/backend/app/analytics/poisson_model.py Normal file
View File

@@ -0,0 +1,113 @@
"""Poisson 分佈賽果預測模組。"""
from __future__ import annotations
import numpy as np
from scipy.stats import poisson
class PoissonMatchPredictor:
"""基於攻守強度的雙方進球機率預測器。"""
def __init__(
self,
home_attack_strength: float,
home_defense_strength: float,
away_attack_strength: float,
away_defense_strength: float,
league_avg_home_goals: float,
) -> None:
for value, name in [
(home_attack_strength, 'home_attack_strength'),
(home_defense_strength, 'home_defense_strength'),
(away_attack_strength, 'away_attack_strength'),
(away_defense_strength, 'away_defense_strength'),
(league_avg_home_goals, 'league_avg_home_goals'),
]:
if value <= 0:
raise ValueError(f'{name} 必須大於 0')
self.home_attack_strength = float(home_attack_strength)
self.home_defense_strength = float(home_defense_strength)
self.away_attack_strength = float(away_attack_strength)
self.away_defense_strength = float(away_defense_strength)
self.league_avg_home_goals = float(league_avg_home_goals)
def calculate_expected_goals(self) -> tuple[float, float]:
"""根據攻守強度與聯盟均值估算預期進球數（λ 值）。
使用比值校正避免極端值放大風險：
- 主隊 λ = 聯盟主場均值 × (主攻 / 客守)
- 客隊 λ = 聯盟客場均值 × (客攻 / 主守)
"""
league_avg_away_goals = self.league_avg_home_goals * 0.95
home_lambda = self.league_avg_home_goals * (self.home_attack_strength / self.away_defense_strength)
away_lambda = league_avg_away_goals * (self.away_attack_strength / self.home_defense_strength)
home_lambda = max(0.01, min(home_lambda, 8.0))
away_lambda = max(0.01, min(away_lambda, 8.0))
return home_lambda, away_lambda
def predict_exact_score_matrix(self, max_goals: int = 5) -> np.ndarray:
"""輸出 0~max_goals 間所有比分組合的機率矩陣。
回傳 shape = (max_goals+1, max_goals+1)，
index [i,j] 代表主隊 i 球、客隊 j 球的機率。
"""
if max_goals < 0:
raise ValueError('max_goals 必須大於等於 0')
home_lambda, away_lambda = self.calculate_expected_goals()
goals = np.arange(max_goals + 1)
home_prob = poisson.pmf(goals, home_lambda)
away_prob = poisson.pmf(goals, away_lambda)
matrix = np.outer(home_prob, away_prob)
matrix = matrix.astype(float)
matrix /= matrix.sum() if matrix.sum() > 0 else 1.0
return matrix
def predict_1x2_probabilities(self) -> dict[str, float]:
"""由波膽矩陣匯總 1x2（主勝/平/客勝）機率。"""
matrix = self.predict_exact_score_matrix(max_goals=8)
draw = float(np.trace(matrix))
home_win = float(np.tril(matrix, -1).sum())
away_win = float(np.triu(matrix, 1).sum())
total = home_win + draw + away_win
if total <= 0:
return {'home_win': 0.0, 'draw': 0.0, 'away_win': 0.0}
return {
'home_win': home_win / total,
'draw': draw / total,
'away_win': away_win / total,
}
def predict_over_under_prob(self, line: float = 2.5, max_goals: int = 8) -> tuple[float, float]:
"""回傳（Under 機率, Over 機率）。"""
if line < 0:
raise ValueError('line 必須大於等於 0')
matrix = self.predict_exact_score_matrix(max_goals=max_goals)
goals = np.arange(max_goals + 1)
home, away = np.meshgrid(goals, goals)
total = home + away
under_mask = total <= line
under = float(matrix[under_mask].sum())
over = float(matrix[~under_mask].sum())
normalizer = under + over
if normalizer <= 0:
return 0.0, 0.0
return under / normalizer, over / normalizer

241
platform/backend/app/analytics/portfolio_analyzer.py Normal file
View File

@@ -0,0 +1,241 @@
"""個人投注弱點分析（Betting Leaks）引擎。
將使用者歷史注單做群組化彙總，找出長期導致虧損的下注模式。
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
def _safe_float(value: Any, default: float | None = None) -> float | None:
try:
return float(value)
except (TypeError, ValueError):
return default
def _safe_int(value: Any, default: int | None = None) -> int | None:
try:
return int(value)
except (TypeError, ValueError):
return default
def _to_bool(value: Any, default: bool = False) -> bool:
if isinstance(value, bool):
return value
if isinstance(value, str):
return value.strip().lower() in {'1', 'true', 't', 'yes', 'y'}
if isinstance(value, (int, float)):
return value not in {0}
return default
def _odds_bucket(odds: float | None, step: float = 0.5) -> str:
if odds is None or odds <= 0:
return 'N/A'
if odds <= 1:
return '1.00-1.50'
bucket_start = ((odds - 1) // step) * step + 1
bucket_end = bucket_start + step
return f'{bucket_start:.2f}-{bucket_end:.2f}'
def _calculate_pnl(stake: float, is_win: bool, closing_odds: float | None, recommended_odds: float | None) -> float:
"""依下注結果與收盤賠率計算實際 P/L。"""
effective_odds = closing_odds
if effective_odds is None or effective_odds <= 1:
effective_odds = recommended_odds
if effective_odds is None or effective_odds <= 1 or stake <= 0:
return 0.0
if is_win:
return stake * (effective_odds - 1)
return -stake
def _calculate_clv(recommended_odds: float | None, closing_odds: float | None) -> float | None:
if recommended_odds is None or closing_odds is None:
return None
if recommended_odds <= 0 or closing_odds <= 0:
return None
return (recommended_odds / closing_odds - 1) * 100
@dataclass(frozen=True)
class LeakageCluster:
market_type: str
bet_type: str
odds_bucket: str
match_stage: str
bet_count: int
total_stake: float
closed_count: int
win_count: int
total_pnl: float
avg_clv_percent: float
roi_percent: float
hit_rate_percent: float
status: str
def as_dict(self) -> dict[str, Any]:
return {
'market_type': self.market_type,
'bet_type': self.bet_type,
'odds_bucket': self.odds_bucket,
'match_stage': self.match_stage,
'bet_count': self.bet_count,
'total_stake': self.total_stake,
'closed_count': self.closed_count,
'win_count': self.win_count,
'total_pnl': self.total_pnl,
'avg_clv_percent': self.avg_clv_percent,
'roi_percent': self.roi_percent,
'hit_rate_percent': self.hit_rate_percent,
'status': self.status,
}
@dataclass(frozen=True)
class HardTruth:
title: str
message: str
cluster: dict[str, Any]
def analyze_user_leaks(user_bets: list[dict[str, Any]]) -> dict[str, Any]:
"""分析使用者注單中的高頻虧損模式，回傳風險群組與漏點警告。"""
raw_bets = user_bets if isinstance(user_bets, list) else []
grouped: dict[tuple[str, str, str, str], dict[str, Any]] = {}
for raw in raw_bets:
if not isinstance(raw, dict):
continue
market_type = str(raw.get('market_type', 'unknown')).strip() or 'unknown'
is_single = raw.get('parlay_type') in (None, 'single', '', 'single_bet')
bet_type = 'single' if is_single else 'parlay'
odds = _safe_float(raw.get('odds'))
stake = _safe_float(raw.get('stake'))
if stake is None or stake <= 0:
continue
match_stage = str(raw.get('match_stage', raw.get('stage', 'unknown'))).strip() or 'unknown'
odds_band = _odds_bucket(odds)
key = (market_type, bet_type, odds_band, match_stage)
entry = grouped.setdefault(
key,
{
'bet_count': 0,
'total_stake': 0.0,
'closed_count': 0,
'win_count': 0,
'total_pnl': 0.0,
'clv_values': [] as list[float],
},
)
entry['bet_count'] += 1
entry['total_stake'] += stake
is_settled = _to_bool(raw.get('is_settled'), default=False)
if not is_settled:
continue
is_win = _to_bool(raw.get('is_win'))
if is_win:
entry['win_count'] += 1
entry['closed_count'] += 1
closing_odds = _safe_float(raw.get('closing_odds'))
recommended_odds = odds or _safe_float(raw.get('recommended_odds'))
pnl = _calculate_pnl(
stake=stake,
is_win=is_win,
closing_odds=closing_odds,
recommended_odds=recommended_odds,
)
entry['total_pnl'] += pnl
clv = _calculate_clv(recommended_odds, closing_odds)
if clv is not None:
entry['clv_values'].append(clv)
total_bets = sum(v['bet_count'] for v in grouped.values())
settled_bets = sum(v['closed_count'] for v in grouped.values())
total_stake = sum(v['total_stake'] for v in grouped.values())
total_pnl = sum(v['total_pnl'] for v in grouped.values())
total_win = sum(v['win_count'] for v in grouped.values())
clusters: list[LeakageCluster] = []
hard_truths: list[HardTruth] = []
for (market_type, bet_type, odds_bucket, match_stage), row in grouped.items():
bet_count = int(row['bet_count'])
closed_count = int(row['closed_count'])
total_stake_group = float(row['total_stake'])
total_pnl_group = float(row['total_pnl'])
roi = (total_pnl_group / total_stake_group * 100) if total_stake_group > 0 else 0.0
win_rate = (row['win_count'] / closed_count * 100) if closed_count > 0 else 0.0
avg_clv = (sum(row['clv_values']) / len(row['clv_values'])) if row['clv_values'] else 0.0
status = 'OK'
if bet_count > 20 and roi < -10:
status = 'CRITICAL_LEAK'
hard_truths.append(
HardTruth(
title='嚴重漏財點',
message=(
f'{match_stage} / {bet_type} / {market_type} / {odds_bucket} 的下注次數 {bet_count} 場，'
f'ROI {roi:.2f}%，請先降低此區塊投注比例。'
),
cluster={
'market_type': market_type,
'bet_type': bet_type,
'odds_bucket': odds_bucket,
'match_stage': match_stage,
},
).__dict__,
)
clusters.append(
LeakageCluster(
market_type=market_type,
bet_type=bet_type,
odds_bucket=odds_bucket,
match_stage=match_stage,
bet_count=bet_count,
total_stake=round(total_stake_group, 2),
closed_count=closed_count,
win_count=row['win_count'],
total_pnl=round(total_pnl_group, 2),
avg_clv_percent=round(avg_clv, 4),
roi_percent=round(roi, 4),
hit_rate_percent=round(win_rate, 2),
status=status,
),
)
clusters.sort(key=lambda c: c.roi_percent)
overall_roi = (total_pnl / total_stake * 100) if total_stake > 0 else 0.0
overall_hit_rate = (total_win / settled_bets * 100) if settled_bets > 0 else 0.0
return {
'total_bet_count': total_bets,
'settled_bet_count': settled_bets,
'total_stake': round(total_stake, 2),
'total_pnl': round(total_pnl, 2),
'overall_roi_percent': round(overall_roi, 4),
'overall_hit_rate_percent': round(overall_hit_rate, 2),
'clusters': [c.as_dict() for c in clusters],
'hard_truths': [h.__dict__ for h in hard_truths],
}

162
platform/backend/app/analytics/proof_of_yield.py Normal file
View File

@@ -0,0 +1,162 @@
"""公開獲利帳本（Proof of Yield）模組。"""
from __future__ import annotations
from dataclasses import dataclass
import json
from pathlib import Path
from typing import Any
from uuid import uuid4
from datetime import datetime
def _as_float(value: Any, *, default: float = 0.0) -> float:
try:
return float(value)
except (TypeError, ValueError):
return default
@dataclass(frozen=True)
class ProofYieldRecord:
recommendation_id: str
match_id: str
market_type: str
selection: str
stake: float
recommended_odds: float
closing_odds: float | None
is_win: bool
settled_at: str
clv_ratio: float | None
clv_percent: float | None
pnl: float
created_at: str
def compute_clv(recommended_odds: float, closing_odds: float) -> float:
"""CLV = (推薦賠率 / 收盤賠率) - 1。"""
if recommended_odds <= 0 or closing_odds <= 0:
raise ValueError('推薦賠率與收盤賠率都必須大於 0')
return (recommended_odds / closing_odds) - 1
def compute_pnl(stake: float, is_win: bool, closing_odds: float | None) -> float:
if closing_odds is None or stake <= 0:
return 0.0
return stake * (closing_odds - 1) if is_win else -stake
@dataclass(frozen=True)
class LedgerSummary:
total_recommendations: int
hit_count: int
win_rate_percent: float
total_stake: float
total_pnl: float
roi_percent: float
avg_clv_percent: float
class ProofOfYieldStore:
"""本地持久化透明帳本（先以 JSON 做可追溯快啟動）。"""
def __init__(self, file_path: str | None = None) -> None:
self.path = Path(file_path or 'data/proof_of_yield_ledger.json')
self.path.parent.mkdir(parents=True, exist_ok=True)
def _load(self) -> list[dict[str, Any]]:
if not self.path.exists():
return []
raw = self.path.read_text(encoding='utf-8')
if not raw.strip():
return []
parsed = json.loads(raw)
if not isinstance(parsed, list):
return []
return parsed
def _save(self, rows: list[dict[str, Any]]) -> None:
self.path.write_text(json.dumps(rows, ensure_ascii=False, indent=2), encoding='utf-8')
def upsert_settlements(self, items: list[dict[str, Any]]) -> list[ProofYieldRecord]:
current = self._load()
idx = {row['recommendation_id']: i for i, row in enumerate(current)}
for item in items:
recommendation_id = str(item.get('recommendation_id') or uuid4().hex)
stake = _as_float(item.get('stake'), default=100.0)
recommended_odds = _as_float(item.get('recommended_odds'))
closing_odds = item.get('closing_odds')
is_win = bool(item.get('is_win', False))
closing = _as_float(closing_odds) if closing_odds is not None else None
clv = None
clv_pct = None
if closing is not None and recommended_odds > 0:
clv = compute_clv(recommended_odds, closing)
clv_pct = clv * 100
pnl = compute_pnl(stake, is_win, closing)
record = {
'recommendation_id': recommendation_id,
'match_id': str(item.get('match_id', 'UNKNOWN')),
'market_type': str(item.get('market_type', '1x2')),
'selection': str(item.get('selection', 'home')),
'stake': round(stake, 4),
'recommended_odds': round(recommended_odds, 6),
'closing_odds': round(closing, 6) if closing is not None else None,
'is_win': is_win,
'settled_at': str(item.get('settled_at') or datetime.utcnow().isoformat()),
'clv_ratio': round(clv, 6) if clv is not None else None,
'clv_percent': round(clv_pct, 4) if clv_pct is not None else None,
'pnl': round(pnl, 4),
'created_at': str(item.get('created_at') or datetime.utcnow().isoformat()),
}
if recommendation_id in idx:
current[idx[recommendation_id]] = record
else:
current.append(record)
self._save(current)
return [ProofYieldRecord(**row) for row in current]
def query_ledger(self, *, limit: int = 200) -> list[ProofYieldRecord]:
rows = sorted(self._load(), key=lambda row: row.get('created_at', ''), reverse=True)
return [ProofYieldRecord(**row) for row in rows[:limit]]
@staticmethod
def summarize(records: list[ProofYieldRecord]) -> LedgerSummary:
total = len(records)
if total == 0:
return LedgerSummary(
total_recommendations=0,
hit_count=0,
win_rate_percent=0.0,
total_stake=0.0,
total_pnl=0.0,
roi_percent=0.0,
avg_clv_percent=0.0,
)
hit = sum(1 for row in records if row.is_win)
total_stake = sum(row.stake for row in records)
total_pnl = sum(row.pnl for row in records)
clv_values = [row.clv_percent for row in records if row.clv_percent is not None]
avg_clv = sum(clv_values) / len(clv_values) if clv_values else 0.0
roi = (total_pnl / total_stake) * 100 if total_stake > 0 else 0.0
win_rate = (hit / total) * 100
return LedgerSummary(
total_recommendations=total,
hit_count=hit,
win_rate_percent=round(win_rate, 4),
total_stake=round(total_stake, 4),
total_pnl=round(total_pnl, 4),
roi_percent=round(roi, 4),
avg_clv_percent=round(avg_clv, 4),
)

53
platform/backend/app/analytics/referee_analyzer.py Normal file
View File

@@ -0,0 +1,53 @@
"""裁判尺度分析器。"""
from __future__ import annotations
from typing import Dict
from .ev_calculator import calculate_expected_value
def calculate_cards_ev(
referee_stats: dict,
match_tension_index: float,
bookmaker_card_line: float,
bookmaker_odds: float,
) -> dict[str, float | bool | str]:
"""判斷裁判/對手張力對紅黃牌盤口的偏差與價值。
依據裁判最近場次平均黃牌數與比賽張力（衝突度）估算
本場真實牌數，並與莊家 O/U 盤口比較。
"""
if bookmaker_odds <= 1:
raise ValueError('bookmaker_odds 必須大於 1')
if bookmaker_card_line <= 0:
raise ValueError('bookmaker_card_line 必須大於 0')
if not 0 <= match_tension_index <= 1:
raise ValueError('match_tension_index 必須在 0~1')
avg_cards = float(referee_stats.get('avg_yellow_cards', 0.0) or 0.0)
penalties_per_game = float(referee_stats.get('penalties_per_game', 0.0) or 0.0)
strictness_index = 20.0 + avg_cards * 1.9 + penalties_per_game * 2.5
# 綜合壓力補正，將裁判嚴厲度與球隊/賽事張力轉為預測牌數。
expected_cards = max(
0.5,
strictness_index * (0.45 + 0.55 * max(0.0, min(match_tension_index, 1.0))),
)
true_prob = min(1.0, max(0.0, expected_cards / (bookmaker_card_line * 1.4)))
implied_prob = 1.0 / bookmaker_odds
edge = true_prob - implied_prob
ev = calculate_expected_value(true_prob, bookmaker_odds, stake=100.0)
return {
'strictness_index': round(strictness_index, 3),
'expected_total_cards': round(expected_cards, 3),
'true_prob': round(true_prob, 4),
'implied_prob': round(implied_prob, 4),
'edge_percent': round(edge * 100, 3),
'is_value_bet': ev['is_value_bet'],
'ev_percentage': ev['ev_percentage'],
}

131
platform/backend/app/analytics/referee_weather.py Normal file
View File

@@ -0,0 +1,131 @@
"""裁判與天候條件量化模組。"""
from __future__ import annotations
from dataclasses import dataclass
def calculate_referee_strictness_index(
avg_yellow_cards: float,
penalties_per_game: float,
) -> float:
"""裁判嚴厲度指標（0-100）。"""
yellow = max(0.0, min(avg_yellow_cards, 8.0)) / 8.0
penalties = max(0.0, min(penalties_per_game, 2.5)) / 2.5
return round(yellow * 55 + penalties * 45, 4)
def detect_cards_pressure_signal(
strictness_index: float,
cards_ou_line: float,
) -> bool:
"""當裁判嚴格且莊家的卡數 O/U 開得偏低時，判斷為可能的逆風盤口。"""
return strictness_index >= 80 and cards_ou_line <= 4.5
def estimate_heat_index(ambient_temp_c: float, humidity_pct: float) -> float:
"""簡化的 Heat Index（攝氏）。"""
t = max(-60.0, min(60.0, ambient_temp_c))
rh = max(0.0, min(100.0, humidity_pct))
hi = (
-8.784695
+ 1.61139411 * t
+ 2.338549 * rh
- 0.14611605 * t * rh
- 0.012308094 * t * t
- 0.016424828 * rh * rh
+ 0.002211732 * t * t * rh
+ 0.00072546 * t * rh * rh
- 0.000003582 * t * t * rh * rh
)
return round(max(0.0, hi), 4)
@dataclass(frozen=True)
class MatchConditionSignal:
strictness_index: float
heat_index: float
cards_pressure_alert: bool
cards_ou_line: float
second_half_home_attack: float
second_half_away_attack: float
second_half_under_recommendation: bool
attacker_direction: str
def adjust_attack_for_heat_and_altitude(
base_attack: float,
*,
heat_index: float,
is_second_half: bool,
venue_altitude_meters: float | None = None,
) -> float:
"""極端環境下的下半場攻擊效率修正。"""
if not is_second_half:
return round(float(base_attack), 6)
heat_penalty = max(0.0, heat_index - 28.0) / 120.0 # 每 1.2 度約降 1%
altitude_penalty = 0.0
if venue_altitude_meters and venue_altitude_meters > 1500:
altitude_penalty = min(0.22, (venue_altitude_meters - 1500) / 8000.0)
factor = max(0.6, 1 - heat_penalty - altitude_penalty)
return round(float(base_attack * factor), 6)
def evaluate_match_conditions(
*,
avg_yellow_cards: float,
penalties_per_game: float,
cards_ou_line: float,
temp_c: float,
humidity_pct: float,
venue_altitude_meters: int,
home_second_half_attack: float,
away_second_half_attack: float,
) -> MatchConditionSignal:
"""整合裁判與天候對下半場盤口與進攻效率的衝擊。"""
strictness_index = calculate_referee_strictness_index(avg_yellow_cards, penalties_per_game)
heat_index = estimate_heat_index(temp_c, humidity_pct)
adjusted_home = adjust_attack_for_heat_and_altitude(
home_second_half_attack,
heat_index=heat_index,
is_second_half=True,
venue_altitude_meters=venue_altitude_meters,
)
adjusted_away = adjust_attack_for_heat_and_altitude(
away_second_half_attack,
heat_index=heat_index,
is_second_half=True,
venue_altitude_meters=venue_altitude_meters,
)
cards_pressure = detect_cards_pressure_signal(strictness_index, cards_ou_line)
high_heat = heat_index >= 32.0
heat_pressure_delta = home_second_half_attack + away_second_half_attack
second_half_under = high_heat and (adjusted_home + adjusted_away) <= heat_pressure_delta * 0.95
if adjusted_home > adjusted_away:
attacker_direction = '上場勢優勢偏向主隊'
elif adjusted_home < adjusted_away:
attacker_direction = '上場勢優勢偏向客隊'
else:
attacker_direction = '攻勢對稱'
return MatchConditionSignal(
strictness_index=strictness_index,
heat_index=heat_index,
cards_pressure_alert=cards_pressure,
cards_ou_line=cards_ou_line,
second_half_home_attack=adjusted_home,
second_half_away_attack=adjusted_away,
second_half_under_recommendation=second_half_under,
attacker_direction=attacker_direction,
)

70
platform/backend/app/analytics/rlm.py Normal file
View File

@@ -0,0 +1,70 @@
"""反向盤口移動（RLM）偵測模組。"""
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime
@dataclass(frozen=True)
class ReverseLineMovementAlert:
match_id: str
market_type: str
selection: str
opening_odds: float
current_odds: float
ticket_pct: float
handle_pct: float
odds_change_pct: float
smart_money_to: str
is_triggered: bool
triggered_at: datetime
rationale: str
def evaluate_reverse_line_movement(
match_id: str,
market_type: str,
selection: str,
*,
opening_odds: float,
current_odds: float,
ticket_pct: float,
handle_pct: float,
ticket_threshold: float = 70.0,
odds_change_threshold: float = 0.05,
) -> ReverseLineMovementAlert:
"""依條件判斷是否出現反向盤口。"""
if opening_odds <= 0:
odds_pct = 0.0
else:
odds_pct = round((current_odds - opening_odds) / opening_odds, 6)
is_triggered = (
ticket_pct > ticket_threshold
and odds_pct > odds_change_threshold
and handle_pct < ticket_pct
)
smart_money_to = selection if handle_pct > ticket_pct else '對側'
rationale = (
f'散戶 {ticket_pct:.1f}% 追捧卻資金 {handle_pct:.1f}%，\n'
f'盤口由 {opening_odds:.2f} 上升到 {current_odds:.2f}'
)
return ReverseLineMovementAlert(
match_id=match_id,
market_type=market_type,
selection=selection,
opening_odds=opening_odds,
current_odds=current_odds,
ticket_pct=ticket_pct,
handle_pct=handle_pct,
odds_change_pct=round(odds_pct * 100, 4),
smart_money_to=smart_money_to,
is_triggered=is_triggered,
triggered_at=datetime.utcnow(),
rationale=rationale,
)

71
platform/backend/app/analytics/sgp_engine.py Normal file
View File

@@ -0,0 +1,71 @@
from typing import List, Dict
import math
class SGPCorrelationEngine:
"""
同場串關 (Same Game Parlay) 關聯性與價值探測引擎
"""
@staticmethod
def calculate_joint_probability(prob_A: float, prob_B: float, correlation_coeff: float) -> float:
"""
計算兩個事件的聯合機率 (考慮相關係數)。
使用簡化的二元正態分佈/Copula近似邏輯。
:param prob_A: 事件 A 獨立發生的真實機率
:param prob_B: 事件 B 獨立發生的真實機率
:param correlation_coeff: 相關係數 (-1.0 到 1.0)
"""
if not (-1.0 <= correlation_coeff <= 1.0):
raise ValueError("相關係數必須介於 -1.0 與 1.0 之間")
# 獨立發生的聯合機率
independent_joint_prob = prob_A * prob_B
# 理論最大與最小邊界
max_joint_prob = min(prob_A, prob_B)
min_joint_prob = max(0.0, prob_A + prob_B - 1.0)
if correlation_coeff == 0:
return independent_joint_prob
elif correlation_coeff > 0:
# 正相關：聯合機率向 max_joint_prob 靠攏
return independent_joint_prob + correlation_coeff * (max_joint_prob - independent_joint_prob)
else:
# 負相關：聯合機率向 min_joint_prob 靠攏
return independent_joint_prob + abs(correlation_coeff) * (min_joint_prob - independent_joint_prob)
@staticmethod
def find_sgp_value(events: List[Dict], bookmaker_sgp_odds: float) -> Dict:
"""
評估 SGP 注單是否具備正期望值。
events 範例: [{'prob': 0.6}, {'prob': 0.4}] 且需自帶兩兩相關係數矩陣 (此處簡化為平均相關性)
"""
if len(events) < 2:
raise ValueError("SGP 必須至少包含兩個事件")
# 假設外部特徵工程已經給出了這組事件的平均正相關係數 (例如 0.4)
# 實務上會透過更複雜的 Monte Carlo 計算，此為展示核心邏輯
avg_correlation = events[0].get('correlation_with_others', 0.0)
current_joint_prob = events[0]['prob']
for i in range(1, len(events)):
current_joint_prob = SGPCorrelationEngine.calculate_joint_probability(
current_joint_prob,
events[i]['prob'],
avg_correlation
)
# 計算莊家隱含機率
implied_prob = 1.0 / bookmaker_sgp_odds
# 計算 EV
ev_percentage = (current_joint_prob * bookmaker_sgp_odds) - 1.0
is_profitable = ev_percentage > 0.05 # 設定 5% 的 EV 門檻
return {
"true_joint_probability": round(current_joint_prob, 4),
"bookmaker_implied_probability": round(implied_prob, 4),
"ev_percentage": round(ev_percentage, 4),
"is_profitable_sgp": is_profitable,
"fair_odds": round(1.0 / current_joint_prob, 2) if current_joint_prob > 0 else 0
}

135
platform/backend/app/analytics/vig_remover.py Normal file
View File

@@ -0,0 +1,135 @@
"""莊家抽水（Vig）去除工具。"""
from __future__ import annotations
from typing import Callable, List, Sequence
import numpy as np
from scipy.optimize import minimize_scalar
def calculate_overround(odds: Sequence[float]) -> float:
"""計算莊家總水位（Overround）。
Overround = Σ(1 / odds_i)。
若結果 > 1 表示含有抽水。
"""
if not odds:
raise ValueError('odds 不可為空')
_odds = np.asarray(odds, dtype=float)
if np.any(_odds <= 1):
raise ValueError('賠率必須全部大於 1')
return float(np.sum(1.0 / _odds))
def remove_margin_basic(odds: Sequence[float]) -> List[float]:
"""等比例剝除抽水。
先轉換為 implied probability，再除以 overround 讓機率總和為 1。
"""
implied = np.array([1.0 / x for x in odds], dtype=float)
overround = implied.sum()
if overround <= 0:
raise ValueError('無效 odds，無法計算去水')
true_probs = implied / overround
return [float(x) for x in true_probs]
def _shin_objective(z: float, observed: np.ndarray) -> float:
"""Shin 模型中，透過 z 估計真實機率，使每個結果有一致修正。
模型假設：
q_i(z) = max((p_i - z/(k-1)) / (1 - k/(k-1)*z), 1e-12)
其中 q_i 為觀察值 implied probability，p_i 為解構後真實機率。
透過約束 Σp_i=1 搜尋最小平方誤差。
"""
k = observed.size
if not 0.0 <= z < 1:
return 1e9
denom = 1.0 - k / max(k - 1, 1) * z
if denom <= 0:
return 1e9
raw = (observed - z / max(k - 1, 1)) / denom
raw = np.clip(raw, 1e-12, None)
normalized = raw / raw.sum()
return float(np.sum((normalized - observed / observed.sum()) ** 2))
def remove_margin_shin(odds: Sequence[float]) -> List[float]:
"""Shin 方法去水。
流程：
1) 觀察賠率轉 implied probability。
2) 用單參數 z 做最小化，推回一組更接近無套利的真實機率。
3) 回傳機率正規化結果。
"""
odds_array = np.asarray(odds, dtype=float)
if odds_array.size == 0:
raise ValueError('odds 不可為空')
if np.any(odds_array <= 1):
raise ValueError('賠率必須全部大於 1')
implied = 1.0 / odds_array
if implied.size == 2:
# 二元市場可直接利用近似閉式解，穩定性較佳
q1 = implied[0] / implied.sum()
q2 = implied[1] / implied.sum()
z = max(0.0, min(0.49, (q1 + q2 - 1.0) * 0.5))
else:
# 多項市場，使用數值搜尋
result = minimize_scalar(
_shin_objective,
args=(implied,),
bounds=(0.0, 0.49),
method='bounded',
)
z = float(result.x if result.success else 0.0)
k = implied.size
denom = 1.0 - k / max(k - 1, 1) * z
if denom <= 0:
return remove_margin_basic(odds)
raw = (implied - z / max(k - 1, 1)) / denom
raw = np.clip(raw, 1e-12, None)
true_prob = raw / raw.sum()
return [float(x) for x in true_prob]
def prob_to_decimal_odds(true_probs: Sequence[float]) -> List[float]:
"""真實機率轉換回無水賠率。
p 轉賠率公式：odds = 1 / p。
"""
probs = np.asarray(true_probs, dtype=float)
if np.any(probs <= 0):
raise ValueError('機率需大於 0')
total = probs.sum()
if not np.isclose(total, 1.0, atol=1e-6):
probs = probs / total
return [round(float(1.0 / p), 4) for p in probs]
def compare_bookmaker_true_prob(
implied_odds: Sequence[float],
transform: Callable[[Sequence[float]], Sequence[float]] = remove_margin_shin,
) -> dict[str, list[float]]:
"""比對原始賠率與去水後真實賠率，可直接提供前端展示。"""
true_probs = transform(implied_odds)
return {
'implied_prob': [float(1.0 / x) for x in implied_odds],
'true_implied_prob': true_probs,
'true_decimal_odds': prob_to_decimal_odds(true_probs),
}