116 lines
3.8 KiB
Python
Executable File
116 lines
3.8 KiB
Python
Executable File
import os
|
|
import json
|
|
import yaml
|
|
from typing import Dict, Any, Optional
|
|
|
|
|
|
class EnsembleConfig:
|
|
_instance: Optional['EnsembleConfig'] = None
|
|
_config: Dict[str, Any] = {}
|
|
|
|
def __new__(cls):
|
|
if cls._instance is None:
|
|
cls._instance = super(EnsembleConfig, cls).__new__(cls)
|
|
cls._instance._load_config()
|
|
return cls._instance
|
|
|
|
def _load_config(self):
|
|
"""Load configuration from YAML file."""
|
|
config_path = os.path.join(os.path.dirname(__file__), 'ensemble_config.yaml')
|
|
try:
|
|
with open(config_path, 'r', encoding='utf-8') as f:
|
|
self._config = yaml.safe_load(f)
|
|
# print(f"✅ Loaded ensemble config from {config_path}")
|
|
except Exception as e:
|
|
print(f"❌ Failed to load ensemble config: {e}")
|
|
self._config = {}
|
|
|
|
def get(self, key: str, default: Any = None) -> Any:
|
|
"""Get configuration value by key (supports dot notation for nested keys)."""
|
|
keys = key.split('.')
|
|
value = self._config
|
|
|
|
try:
|
|
for k in keys:
|
|
value = value[k]
|
|
return value
|
|
except (KeyError, TypeError):
|
|
return default
|
|
|
|
|
|
# Singleton accessor
|
|
def get_config() -> EnsembleConfig:
|
|
return EnsembleConfig()
|
|
|
|
|
|
# ── Market Thresholds Loader ────────────────────────────────────────────
|
|
|
|
_market_thresholds_cache: Optional[Dict[str, Any]] = None
|
|
|
|
|
|
def load_market_thresholds() -> Dict[str, Any]:
|
|
"""
|
|
Load market thresholds from JSON config file.
|
|
Returns the full config dict with 'markets' and 'defaults' keys.
|
|
Caches after first load for performance.
|
|
"""
|
|
global _market_thresholds_cache
|
|
if _market_thresholds_cache is not None:
|
|
return _market_thresholds_cache
|
|
|
|
config_path = os.path.join(os.path.dirname(__file__), 'market_thresholds.json')
|
|
try:
|
|
with open(config_path, 'r', encoding='utf-8') as f:
|
|
data = json.load(f)
|
|
_market_thresholds_cache = data
|
|
print(f"✅ Market thresholds loaded: {len(data.get('markets', {}))} markets (v={data.get('_meta', {}).get('version', '?')})")
|
|
return data
|
|
except Exception as e:
|
|
print(f"❌ Failed to load market thresholds: {e} — using built-in defaults")
|
|
_market_thresholds_cache = {"markets": {}, "defaults": {
|
|
"calibration": 0.55,
|
|
"min_conf": 55.0,
|
|
"min_play_score": 68.0,
|
|
"min_edge": 0.02,
|
|
"odds_band_min_sample": 0.0,
|
|
"odds_band_min_edge": 0.0,
|
|
}}
|
|
return _market_thresholds_cache
|
|
|
|
|
|
def build_threshold_dict(field: str) -> Dict[str, float]:
|
|
"""
|
|
Build a flat {market: value} dict for a specific threshold field.
|
|
|
|
Usage:
|
|
calibration_map = build_threshold_dict("calibration")
|
|
# → {"MS": 0.62, "DC": 0.82, ...}
|
|
"""
|
|
data = load_market_thresholds()
|
|
markets = data.get("markets", {})
|
|
result: Dict[str, float] = {}
|
|
for market, cfg in markets.items():
|
|
if field in cfg:
|
|
result[market] = float(cfg[field])
|
|
return result
|
|
|
|
|
|
def get_threshold_default(field: str) -> float:
|
|
"""Get the default fallback value for a threshold field."""
|
|
data = load_market_thresholds()
|
|
defaults = data.get("defaults", {})
|
|
return float(defaults.get(field, 0.0))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Test
|
|
cfg = get_config()
|
|
print(f"Weights: {cfg.get('engine_weights')}")
|
|
print(f"Team Weight: {cfg.get('engine_weights.team')}")
|
|
print()
|
|
print("--- Market Thresholds ---")
|
|
for field in ["calibration", "min_conf", "min_play_score", "min_edge"]:
|
|
d = build_threshold_dict(field)
|
|
print(f"{field}: {d}")
|
|
print(f"Default calibration: {get_threshold_default('calibration')}")
|