""" Deterministic Buy/Hold/Sell policy applied to Moirai-MoE's forecast output (spec sections 31-34, 93, 111-113). This module contains NO predictive model of its own, hidden or otherwise — only arithmetic over a forecast object produced by moirai_model.py. Nothing in this file imports sklearn/xgboost/lightgbm/catboost or any other classifier. """ from __future__ import annotations from dataclasses import dataclass from typing import Optional import numpy as np import config as cfg from calibration import ProbabilityCalibrator @dataclass class Decision: action: str # "BUY" | "SELL" | "HOLD" expected_return: float expected_edge: float p10_return: float p90_return: float prob_favorable: float # the probability actually used to gate this decision (calibrated, if a calibrator was supplied) raw_prob_favorable: float # the same event's probability BEFORE calibration -- always shown, never hidden (spec: "do not hide weak signals") calibrated: bool # whether a calibrator was actually applied interval_width_pct: float rationale: str hold_reason: str = None reject_stage: str = None # structured category for diagnostics/counting: # "wide_interval" | "no_edge" | "low_confidence_buy" | "low_confidence_sell" | None (BUY/SELL taken) def _prob_exceeds(levels: list, values: np.ndarray, threshold: float) -> float: """P(true return > threshold), estimated by linear interpolation on the model's own empirical quantile function — not a fabricated number (spec section 32-33).""" order = np.argsort(levels) lv = np.asarray(levels)[order] vv = np.asarray(values)[order] if threshold <= vv[0]: return 1.0 if threshold >= vv[-1]: return 0.0 p_below = float(np.interp(threshold, vv, lv)) return 1.0 - p_below def decide(forecast, price_now: float, dc: cfg.DecisionConfig = None, cost_estimate_frac: float = None, calibrator: Optional[ProbabilityCalibrator] = None) -> Decision: """`forecast` is a QuantileForecast (see moirai_model.py) with .quantile_levels (list of floats incl. 0.1/0.5/0.9) and .quantiles of shape (horizon, len(levels)). `calibrator`, if given, is applied to prob_favorable/ prob_favorable_sell BEFORE either is compared against dc.min_probability -- real EURUSD 1h testing showed raw quantile- derived probabilities of 0.70-0.88 against a realized win rate around 0.50, i.e. the raw number is not trustworthy on its own (see calibration.py). None (the default) reproduces the previous, uncalibrated behavior exactly -- calibration is opt-in via an explicitly-supplied, already-fit ProbabilityCalibrator (see walk_forward.py for the only leakage-safe way to obtain one).""" dc = dc or cfg.DecisionConfig() levels = list(forecast.quantile_levels) step_prices = np.asarray(forecast.quantiles)[-1] # final horizon step step_returns = step_prices / price_now - 1.0 p10 = float(step_returns[levels.index(0.1)]) p50 = float(step_returns[levels.index(0.5)]) p90 = float(step_returns[levels.index(0.9)]) cost = cost_estimate_frac if cost_estimate_frac is not None else dc.cost_bps / 10_000 interval_width_pct = p90 - p10 raw_prob_favorable = _prob_exceeds(levels, step_returns, threshold=cost) # P(return > +cost) -- the BUY-side confidence # SELL is profitable when return < -cost (shorting profits from a # price drop), which is NOT the same event as "return <= +cost" -- # that includes the whole middle range between -cost and +cost too, # so it isn't "1 - prob_favorable" reused. This is its own directional # probability, symmetric to prob_favorable but mirrored around # -cost instead of +cost, exactly the way min_probability's own # definition ("P(return favorable) required to act", config.py) is # meant to apply to either direction, not just BUY. raw_prob_favorable_sell = 1.0 - _prob_exceeds(levels, step_returns, threshold=-cost) # P(return < -cost) if calibrator is not None: prob_favorable = calibrator.apply(raw_prob_favorable) prob_favorable_sell = calibrator.apply(raw_prob_favorable_sell) calibrated = True else: prob_favorable = raw_prob_favorable prob_favorable_sell = raw_prob_favorable_sell calibrated = False hold_reason = None reject_stage = None if interval_width_pct > dc.max_interval_width_pct: action = "HOLD" hold_reason = "forecast interval too wide relative to expected move (section 93)" reject_stage = "wide_interval" elif (p50 - cost) > dc.buy_threshold and prob_favorable >= dc.min_probability: action = "BUY" elif (p50 + cost) < dc.sell_threshold and prob_favorable_sell >= dc.min_probability: action = "SELL" else: action = "HOLD" # Distinguish "p50 never even cleared a directional threshold" from # "it cleared one, but confidence didn't clear min_probability" -- # these are very different diagnoses (a fast/low-volatility # timeframe where p50 rarely moves enough to matter, vs. a # timeframe with real moves but noisy/unreliable ones) and were # previously indistinguishable from outside this function, which # made "why are there almost no trades" unanswerable without this # split (spec section 98: errors/holds must say what failed). buy_edge_ok = (p50 - cost) > dc.buy_threshold sell_edge_ok = (p50 + cost) < dc.sell_threshold if buy_edge_ok: hold_reason = "expected edge below threshold after costs — buy-side edge present but confidence below min_probability (section 93)" reject_stage = "low_confidence_buy" elif sell_edge_ok: hold_reason = "expected edge below threshold after costs — sell-side edge present but confidence below min_probability (section 93)" reject_stage = "low_confidence_sell" else: hold_reason = "expected edge below threshold after costs — no directional edge either way (section 93)" reject_stage = "no_edge" # Report whichever probability is actually relevant to the action taken # -- for a SELL, "P(return>cost)" reads as near-zero and looks like the # decision has no confidence behind it, when the real (and high) # confidence is in the mirrored SELL-side event. BUY/HOLD keep the # original P(return>cost) framing. if action == "SELL": prob_shown, raw_prob_shown, prob_label = prob_favorable_sell, raw_prob_favorable_sell, "P(return<-cost)" else: prob_shown, raw_prob_shown, prob_label = prob_favorable, raw_prob_favorable, "P(return>cost)" cal_note = f"; calibrated (raw={raw_prob_shown:.2f})" if calibrated else "" rationale = ( f"P50 return {p50:+.4%} vs cost {cost:.4%}; {prob_label}={prob_shown:.2f}{cal_note}; " f"interval[P10,P90]=[{p10:+.4%},{p90:+.4%}]" ) return Decision( action=action, expected_return=p50, expected_edge=p50 - cost, p10_return=p10, p90_return=p90, prob_favorable=prob_shown, raw_prob_favorable=raw_prob_shown, calibrated=calibrated, interval_width_pct=interval_width_pct, rationale=rationale, hold_reason=hold_reason, reject_stage=reject_stage, )