File size: 7,400 Bytes
ef20ebe | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 | """
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,
)
|