fix: distinguish reference odds coverage in daily cards
All checks were successful
2026 World Cup Quant Platform - Production Deployment / Code Quality, Security Gate & Testing (push) Successful in 4m13s
2026 World Cup Quant Platform - Production Deployment / Deploy to Production VM via Gitea CD (push) Successful in 1m22s

This commit is contained in:
OG T
2026-06-19 00:32:45 +08:00
parent bd2fb5cc33
commit ef15746130

View File

@@ -656,6 +656,49 @@ def _apply_recent_market_calibration(item: dict[str, Any]) -> dict[str, Any]:
return adjusted
REFERENCE_MARKET_SOURCE_KINDS = {
'reference_market',
'single_provider_market',
'mixed_market_source',
}
def _market_source_counts(items: list[dict[str, Any]]) -> dict[str, int]:
formal_count = 0
reference_count = 0
for item in items:
if bool(item.get('has_market_odds')):
formal_count += 1
continue
source_kind = str(item.get('odds_source_kind') or '')
source_label = str(item.get('odds_source_label') or '')
target_odds = _safe_float(item.get('target_odds'), 0.0)
if target_odds > 1 and (
source_kind in REFERENCE_MARKET_SOURCE_KINDS
or '參考盤' in source_label
or '台灣運彩' in source_label
):
reference_count += 1
watchlist_count = max(0, len(items) - formal_count - reference_count)
return {
'formal_market_count': formal_count,
'reference_market_count': reference_count,
'pre_market_watchlist_count': watchlist_count,
'live_market_count': formal_count,
'pre_market_count': len(items) - formal_count,
}
def _market_data_status_from_counts(counts: dict[str, int]) -> str:
if counts.get('formal_market_count', 0) > 0:
return 'live_market_available'
if counts.get('reference_market_count', 0) > 0:
return 'reference_market_watchlist'
return 'pre_market_watchlist'
def _market_tier_from_item(item: dict[str, Any]) -> str:
market_type = str(item.get('market_type') or '')
@@ -736,13 +779,22 @@ def recalibrate_daily_card_confidence_payload(card_payload: dict[str, Any]) -> d
next_group.append(item)
payload[bucket_name] = next_group
if changed_count or market_calibrated_count:
payload_items = [
item
for bucket_name in bucket_names
for item in (payload.get(bucket_name, []) or [])
if isinstance(item, dict)
]
source_counts = _market_source_counts(payload_items)
if changed_count or market_calibrated_count or payload_items:
quality_summary = dict(payload.get('data_quality_summary') or {})
if changed_count:
quality_summary['confidence_recalibrated_items'] = changed_count
if market_calibrated_count:
quality_summary['recent_market_calibrated_count'] = market_calibrated_count
quality_summary.update(source_counts)
payload['data_quality_summary'] = quality_summary
payload['market_data_status'] = _market_data_status_from_counts(source_counts)
if market_calibrated_count:
summary = str(payload.get('summary') or '')
calibration_summary = (
@@ -751,6 +803,14 @@ def recalibrate_daily_card_confidence_payload(card_payload: dict[str, Any]) -> d
)
if '近 7 天賽後玩法校準' not in summary:
payload['summary'] = f'{summary}{calibration_summary}'
if source_counts.get('reference_market_count', 0) > 0 and source_counts.get('formal_market_count', 0) <= 0:
summary = str(payload.get('summary') or '')
reference_summary = (
f' 目前有 {source_counts["reference_market_count"]} 組台灣/單一來源參考盤監控,'
'但尚未達多莊家正式盤口標準,因此只能做賠率監控與進場條件,不可標成正式下注訊號。'
)
if '單一來源參考盤監控' not in summary:
payload['summary'] = f'{summary}{reference_summary}'
return payload
def _confidence_band(score: float, has_market_odds: bool, risk_level: str) -> str:
@@ -1821,8 +1881,12 @@ def generate_daily_card(target_date: str, matches: list[dict[str, Any]]) -> dict
)
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
source_counts = _market_source_counts(all_items)
if source_counts.get('reference_market_count', 0) > 0 and source_counts.get('formal_market_count', 0) <= 0:
summary = (
f'{summary} 目前有 {source_counts["reference_market_count"]} 組台灣/單一來源參考盤監控,'
'但尚未達多莊家正式盤口標準,因此只能做賠率監控與進場條件,不可標成正式下注訊號。'
)
quality_counts: dict[str, int] = {}
for item in all_items:
quality = str(item.get('data_quality') or 'unknown')
@@ -1834,11 +1898,10 @@ def generate_daily_card(target_date: str, matches: list[dict[str, Any]]) -> dict
'total_daily_amount_twd': _stake_amount_twd(total_unit),
'unit_size_twd': DEFAULT_UNIT_STAKE_TWD,
'summary': summary,
'market_data_status': 'live_market_available' if live_market_count > 0 else 'pre_market_watchlist',
'market_data_status': _market_data_status_from_counts(source_counts),
'data_quality_summary': {
**quality_counts,
'live_market_count': live_market_count,
'pre_market_count': pre_market_count,
**source_counts,
'recent_market_calibrated_count': calibrated_count,
},
'execution_policy': (