54 lines
2.9 KiB
Python
54 lines
2.9 KiB
Python
from dataclasses import dataclass
|
|
from datetime import timedelta
|
|
from .domain import default_registry,number,utc,ZURICH
|
|
|
|
@dataclass(frozen=True)
|
|
class ReplayScore:
|
|
family:str
|
|
comparison_key:str
|
|
first_decision:object
|
|
last_decision:object
|
|
forecast_issued_at:object
|
|
actual_available_at:object
|
|
cost_chf:float
|
|
coverage:float
|
|
days:int
|
|
constraint_breaches:int=0
|
|
terminal_normalized:bool=True
|
|
def valid(self):
|
|
number(self.cost_chf,'replay cost');number(self.coverage,'coverage',0,1)
|
|
return (utc(self.forecast_issued_at)<=utc(self.first_decision) and utc(self.actual_available_at)>=utc(self.last_decision) and self.terminal_normalized and self.constraint_breaches==0)
|
|
|
|
def choose_family(setting,current,scores,*,registry=None,lookback_days=14,minimum_days=7,minimum_coverage=.9,margin_chf=1.,now):
|
|
registry=registry or default_registry()
|
|
if setting!='auto':
|
|
registry.get(setting)
|
|
return {'family':setting,'mode':'configured','reason':'Explicit installation setting'}
|
|
registry.get(current);number(margin_chf,'margin',0);valid=[]
|
|
for score in scores:
|
|
registry.get(score.family)
|
|
if (score.valid() and score.coverage>=minimum_coverage and score.days>=minimum_days
|
|
and utc(score.first_decision)>=utc(now)-timedelta(days=lookback_days)
|
|
and utc(score.last_decision)<=utc(now) and utc(score.actual_available_at)<=utc(now)):
|
|
valid.append(score)
|
|
groups={}
|
|
for score in valid:
|
|
key=(score.comparison_key,utc(score.first_decision),utc(score.last_decision),score.days)
|
|
groups.setdefault(key,{})[score.family]=score
|
|
complete=[g for g in groups.values() if len(g)==len(registry.entries())]
|
|
if not complete:return {'family':current,'mode':'collecting','reason':'Insufficient comparable out-of-sample replay evidence'}
|
|
group=max(complete,key=lambda g:utc(next(iter(g.values())).last_decision))
|
|
best=min(group,key=lambda f:(group[f].cost_chf,f));improvement=group[current].cost_chf-group[best].cost_chf
|
|
chosen=best if improvement>margin_chf else current
|
|
return {'family':chosen,'mode':'economic_replay','improvementChf':improvement,'reason':'Matched historical cost replay; switching margin applied','costByFamilyChf':{k:v.cost_chf for k,v in group.items()}}
|
|
|
|
def training_due(last_trained_at,now,cadence='daily'):
|
|
if cadence not in ('daily','weekly'):raise ValueError('Unknown training cadence')
|
|
if last_trained_at is None:return True
|
|
now,last=utc(now).astimezone(ZURICH),utc(last_trained_at).astimezone(ZURICH)
|
|
return (now.date()-last.date()).days >= (1 if cadence=='daily' else 7)
|
|
|
|
def promote_candidate(*,active_cost,candidate_cost,valid_coverage,no_data_leakage,constraints_passed):
|
|
number(active_cost,'active cost');number(candidate_cost,'candidate cost')
|
|
return bool(valid_coverage and no_data_leakage and constraints_passed and candidate_cost<active_cost)
|