fix: expose formal odds provider blockers
Some checks failed
2026 World Cup Quant Platform - Production Deployment / Code Quality, Security Gate & Testing (push) Successful in 3m56s
2026 World Cup Quant Platform - Production Deployment / Deploy to Production VM via Gitea CD (push) Has been cancelled

This commit is contained in:
OG T
2026-06-19 00:44:11 +08:00
parent ef15746130
commit 8d7e34e6bd
2 changed files with 106 additions and 12 deletions

View File

@@ -48,6 +48,7 @@ TAIWAN_SPORTS_LOTTERY_WC_URL = os.environ.get(
'TAIWAN_SPORTS_LOTTERY_WC_URL',
'https://blob3rd.sportslottery.com.tw/apidata/Pre/WC-Games.zh.json',
)
THE_ODDS_PLACEHOLDER_KEYS = {'your_the_odds_api_key', 'changeme', 'placeholder', 'todo'}
FEATURED_MARKET_KEYS = {'h2h', 'spreads', 'totals', 'outrights', 'h2h_lay', 'outrights_lay'}
TEAM_ALIAS_MAP = {
@@ -574,10 +575,7 @@ async def fetch_espn_scoreboard(client: httpx.AsyncClient) -> list[dict[str, Any
end = now + timedelta(days=ESPN_SCOREBOARD_LOOKAHEAD_DAYS)
params = {'dates': f'{start:%Y%m%d}-{end:%Y%m%d}'}
logger.info(
'未設定 THE_ODDS_API_KEY使用 ESPN scoreboard 作為低階備援。dates=%s',
params['dates'],
)
logger.info('同步 ESPN scoreboard 比分備援。dates=%s', params['dates'])
payload = await _request_json(client, ESPN_SCOREBOARD_URL, params=params)
if not isinstance(payload, dict):
logger.warning('ESPN scoreboard 回傳格式不是物件,略過本輪比分同步。')
@@ -726,7 +724,41 @@ async def fetch_taiwan_sports_lottery_reference(client: httpx.AsyncClient) -> li
return parsed
async def fetch_source_payload() -> tuple[str, list[dict[str, Any]]]:
def _primary_provider_status(
*,
status: str | None = None,
message: str | None = None,
event_count: int | None = None,
) -> dict[str, Any]:
key_is_configured = bool(THE_ODDS_API_KEY)
key_is_placeholder = THE_ODDS_API_KEY.lower() in THE_ODDS_PLACEHOLDER_KEYS
if status is None:
if not key_is_configured:
status = 'missing_key'
message = '正式 The Odds API key 尚未設定。'
elif key_is_placeholder:
status = 'placeholder_key'
message = '正式 The Odds API key 仍是 placeholder未接入可用 provider。'
else:
status = 'configured'
message = '正式 The Odds API key 已設定,等待抓取結果確認。'
payload: dict[str, Any] = {
'provider': 'the-odds-api',
'status': status,
'api_key_configured': key_is_configured,
'api_key_placeholder': key_is_placeholder,
'sport_key': THE_ODDS_SPORT_KEY,
'regions': THE_ODDS_REGIONS,
'markets': _market_keys(THE_ODDS_MARKETS),
'message': message,
}
if event_count is not None:
payload['event_count'] = event_count
return payload
async def fetch_source_payload() -> tuple[str, list[dict[str, Any]], dict[str, Any]]:
async with httpx.AsyncClient(
headers={
'User-Agent': 'Mozilla/5.0 FIFA2026QuantBot/1.0',
@@ -734,16 +766,39 @@ async def fetch_source_payload() -> tuple[str, list[dict[str, Any]]]:
'Referer': 'https://www.sportslottery.com.tw/',
}
) as client:
if THE_ODDS_API_KEY and THE_ODDS_API_KEY != 'your_the_odds_api_key':
return 'the-odds-api', await fetch_the_odds_api(client)
primary_status = _primary_provider_status()
if primary_status['api_key_configured'] and not primary_status['api_key_placeholder']:
try:
the_odds_events = await fetch_the_odds_api(client)
if the_odds_events:
return 'the-odds-api', the_odds_events, _primary_provider_status(
status='ok',
message='正式 The Odds API 已回傳可用賽事與盤口。',
event_count=len(the_odds_events),
)
primary_status = _primary_provider_status(
status='empty_events',
message='正式 The Odds API 已可呼叫,但目前沒有回傳可用世界盃盤口;暫時回退台灣參考盤。',
event_count=0,
)
except Exception as exc:
primary_status = _primary_provider_status(
status='error',
message=f'正式 The Odds API 抓取失敗,已回退台灣參考盤:{str(exc)[:220]}',
event_count=0,
)
logger.warning('The Odds API 抓取失敗,回退台灣運彩參考盤:%s', exc)
else:
logger.warning('The Odds API 未可用:%s', primary_status.get('message'))
try:
taiwan_reference_events = await fetch_taiwan_sports_lottery_reference(client)
except Exception as exc:
logger.warning('台灣運彩參考盤抓取失敗,本輪僅使用 ESPN 備援:%s', exc)
taiwan_reference_events = []
if taiwan_reference_events:
return 'taiwan-sports-lottery-reference', taiwan_reference_events
return 'espn-scoreboard', await fetch_espn_scoreboard(client)
return 'taiwan-sports-lottery-reference', taiwan_reference_events, primary_status
return 'espn-scoreboard', await fetch_espn_scoreboard(client), primary_status
async def process_odds_data(data: list[dict[str, Any]]) -> dict[str, int]:
@@ -946,7 +1001,7 @@ async def publish_status(status: dict[str, Any]) -> None:
async def run_once() -> dict[str, Any]:
source, payload = await fetch_source_payload()
source, payload, primary_provider = await fetch_source_payload()
stats = await process_odds_data(payload)
if source != 'espn-scoreboard':
try:
@@ -974,6 +1029,7 @@ async def run_once() -> dict[str, Any]:
'status': 'ok',
'source': source,
'run_at': datetime.now(timezone.utc).isoformat(),
'primary_provider': primary_provider,
**stats,
}
await publish_status(status)
@@ -991,6 +1047,7 @@ async def run_forever() -> None:
'status': 'error',
'source': 'unknown',
'run_at': datetime.now(timezone.utc).isoformat(),
'primary_provider': _primary_provider_status(status='unknown', message='本輪 ingestion 失敗,未能確認正式 provider 狀態。'),
'message': str(exc),
}
await publish_status(error_status)

View File

@@ -827,6 +827,28 @@ async def analytics_source_health() -> SourceHealthResponse:
upcoming_odds_matches = int(upcoming_odds_result.scalar_one() or 0)
worker_status = str((ingestion_status or {}).get('status') or 'unknown')
source_name = str((ingestion_status or {}).get('source') or 'unknown')
primary_provider_status = (ingestion_status or {}).get('primary_provider')
if not isinstance(primary_provider_status, dict):
primary_provider_status = {
'provider': 'the-odds-api',
'status': 'unknown',
'api_key_configured': False,
'api_key_placeholder': False,
'message': '尚未收到 odds worker 的正式 provider 狀態回報。',
}
primary_provider_state = str(primary_provider_status.get('status') or 'unknown')
if primary_provider_state == 'ok':
formal_provider_blocker = None
elif primary_provider_state == 'placeholder_key':
formal_provider_blocker = 'THE_ODDS_API_KEY 仍是 placeholder請改用有效正式 key 後才可升級多來源盤口。'
elif primary_provider_state == 'missing_key':
formal_provider_blocker = 'THE_ODDS_API_KEY 尚未設定,無法抓取正式多來源盤口。'
elif primary_provider_state == 'empty_events':
formal_provider_blocker = 'The Odds API 可呼叫,但目前沒有回傳可用世界盃盤口;暫以台灣參考盤監控。'
elif primary_provider_state == 'error':
formal_provider_blocker = str(primary_provider_status.get('message') or 'The Odds API 最近一次抓取失敗。')
else:
formal_provider_blocker = '尚未確認 The Odds API 正式 provider 狀態。'
if odds_rows <= 0 or latest_odds is None or worker_status == 'error':
odds_coverage_status = 'stale'
elif 'taiwan-sports-lottery-reference' in source_name:
@@ -873,9 +895,11 @@ async def analytics_source_health() -> SourceHealthResponse:
'alternate_team_totals',
],
'derived_recommendation_markets': ['double_chance'],
'formal_provider_status': primary_provider_status,
'formal_provider_blocker': formal_provider_blocker,
'taiwan_sports_lottery': '已確認公開世界盃參考盤端點 Pre/WC-Games.zh.json目前定位為台灣盤比對與最低可接受賠率參考不等同多莊家正式 provider。',
'taiwan_sports_lottery_status': 'reference_market_enabled' if 'taiwan-sports-lottery-reference' in source_name else 'reference_market_waiting',
'current_limitation': '目前若未接入多莊家正式 odds provider未來賽事只能產生台灣盤參考與預掛條件不能包裝成保證高勝率。',
'current_limitation': formal_provider_blocker or '已接入正式 provider仍需檢查每個玩法至少兩家莊家與未來賽事覆蓋率。',
},
)
@@ -1056,13 +1080,25 @@ async def analytics_recommendation_readiness(days_ahead: int = 2) -> dict[str, A
source_name = str((ingestion_status or {}).get('source') or 'unknown')
worker_status = str((ingestion_status or {}).get('status') or 'unknown')
primary_provider_status = (ingestion_status or {}).get('primary_provider')
if not isinstance(primary_provider_status, dict):
primary_provider_status = {}
primary_provider_state = str(primary_provider_status.get('status') or 'unknown')
blocking_reasons: list[str] = []
warnings: list[str] = []
if upcoming_match_count <= 0:
blocking_reasons.append('未來視窗內沒有可分析賽事。')
if source_name != 'the-odds-api':
if 'taiwan-sports-lottery-reference' in source_name:
if primary_provider_state == 'placeholder_key':
blocking_reasons.append('THE_ODDS_API_KEY 仍是 placeholder尚未接入有效正式 odds provider。')
elif primary_provider_state == 'missing_key':
blocking_reasons.append('THE_ODDS_API_KEY 尚未設定,無法抓取正式多來源盤口。')
elif primary_provider_state == 'empty_events':
blocking_reasons.append('The Odds API 已可呼叫,但目前沒有回傳可用世界盃盤口;只能暫時使用參考盤監控。')
elif primary_provider_state == 'error':
blocking_reasons.append(str(primary_provider_status.get('message') or 'The Odds API 最近一次抓取失敗。'))
elif 'taiwan-sports-lottery-reference' in source_name:
blocking_reasons.append('目前包含台灣運彩參考盤,但仍不是多莊家正式 odds provider只能作為最低賠率比對與預掛觀察。')
else:
blocking_reasons.append('目前賠率來源不是正式 odds provider只能使用比分備援與預掛觀察。')
@@ -1111,6 +1147,7 @@ async def analytics_recommendation_readiness(days_ahead: int = 2) -> dict[str, A
'name': source_name,
'worker_status': worker_status,
'last_run': ingestion_status,
'formal_provider_status': primary_provider_status,
},
'thresholds': {
'minimum_bookmakers': minimum_bookmakers,