sevenhill/TradeBot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: train forecasts on trade outcomes

Browse Source
This commit is contained in:
Курнат Андрей
2026-07-14 07:49:32 +03:00
parent 668e606ee2
commit 7186acb9a1
18 changed files with 867 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crypto_spot_bot/time_series.py
+138 -9
View File
@@ -163,6 +163,7 @@ class TimeSeriesForecast:
calibrated_min_edge_percent: float = 0.0
calibrated_min_probability_up: float = 0.0
calibrated_min_confidence: float = 0.0
probability_take_profit_first: float | None = None
def as_dict(self) -> dict[str, Any]:
return asdict(self)
@@ -254,6 +255,7 @@ class TimeSeriesForecaster:
expected_gross_return = float(selected.get("expected_gross_return", expected_return))
expected_price = closes[-1] * math.exp(expected_gross_return)
probability_up = _clamp(float(selected.get("probability_up", 0.5)), 0.0, 1.0)
target_transform = str(entry.get("target_transform", "net_return_over_volatility"))
model_mae = max(float(selected.get("validation_mae", 0.0)), 1e-9)
baseline_mae = max(float(selected.get("baseline_mae", model_mae)), model_mae)
uncertainty = max(float(selected.get("uncertainty", model_mae)), 1e-9)
@@ -279,12 +281,16 @@ class TimeSeriesForecaster:
or (expected_return_percent <= -min_edge and probability_up <= 0.45)
or (q50_percent <= -min_edge and probability_up <= 0.48)
)
reason = _reason(
model=model,
expected_return_percent=expected_return_percent,
probability_up=probability_up,
skill=skill,
block_entry=block_entry,
reason = (
_barrier_reason(model, expected_return_percent, probability_up, skill, block_entry)
if target_transform == "barrier_net_return"
else _reason(
model=model,
expected_return_percent=expected_return_percent,
probability_up=probability_up,
skill=skill,
block_entry=block_entry,
)
)
if not symbol_eligible:
reason = "symbol excluded by train-only calibration"
@@ -292,7 +298,11 @@ class TimeSeriesForecaster:
enabled=True,
usable=True,
model=model,
volatility_model="probabilistic multi-horizon after-cost quantile",
volatility_model=(
"TP-before-SL multi-task after-cost model"
if target_transform == "barrier_net_return"
else "probabilistic multi-horizon after-cost quantile"
),
expected_return_percent=round(expected_return_percent, 4),
expected_price=round(expected_price, 8),
volatility_percent=round(volatility_percent, 4),
@@ -309,7 +319,7 @@ class TimeSeriesForecaster:
quantile_50_percent=round(q50_percent, 4),
quantile_90_percent=round(q90_percent, 4),
conservative_return_percent=round(conservative_return_percent, 4),
target_transform=str(entry.get("target_transform", "net_return_over_volatility")),
target_transform=target_transform,
feature_snapshot=feature_snapshot,
horizon_forecasts=_public_horizon_forecasts(prediction),
candidates=[{"model": model, "mae_percent": round(model_mae * 100, 4)}],
@@ -321,6 +331,11 @@ class TimeSeriesForecaster:
calibrated_min_edge_percent=calibrated["edge"],
calibrated_min_probability_up=calibrated["probability"],
calibrated_min_confidence=calibrated["confidence"],
probability_take_profit_first=(
round(probability_up, 4)
if target_transform == "barrier_net_return"
else None
),
)
direct_horizon = _is_direct_horizon(entry)
@@ -1172,6 +1187,47 @@ def _average_ensemble_predictions(predictions: list[float | dict[str, Any]]) ->
def _torch_head_outputs(context: list[float], entry: dict[str, Any], hidden_size: int) -> list[float]:
context = _apply_context_norm(context, entry)
if entry.get("multitask_head") is True:
hidden_matrix = _float_matrix(entry.get("head_hidden_weight"))
hidden_bias = _float_vector(entry.get("head_hidden_bias"))
if not hidden_matrix or len(hidden_bias) != len(hidden_matrix):
return []
shared = [
_gelu(_dot(row, context) + hidden_bias[index])
for index, row in enumerate(hidden_matrix)
if len(row) == hidden_size
]
if len(shared) != len(hidden_matrix):
return []
return_matrix = _float_matrix(entry.get("return_head_weight"))
return_bias = _float_vector(entry.get("return_head_bias"))
event_matrix = _float_matrix(entry.get("event_head_weight"))
event_bias = _float_vector(entry.get("event_head_bias"))
if (
not return_matrix
or len(return_bias) != len(return_matrix)
or not event_matrix
or len(event_bias) != len(event_matrix)
):
return []
return_values = [
_dot(row, shared) + return_bias[index]
for index, row in enumerate(return_matrix)
if len(row) == len(shared)
]
event_values = [
_dot(row, shared) + event_bias[index]
for index, row in enumerate(event_matrix)
if len(row) == len(shared)
]
if len(return_values) != len(event_values) * 4:
return []
outputs: list[float] = []
for horizon_index, event_value in enumerate(event_values):
base = horizon_index * 4
outputs.extend(return_values[base : base + 4])
outputs.append(event_value)
return outputs
raw_weight = entry.get("head_weight")
if isinstance(raw_weight, list) and raw_weight and isinstance(raw_weight[0], list):
matrix = _float_matrix(raw_weight)
@@ -1242,8 +1298,18 @@ def _decode_multi_horizon_prediction(
expected = decode("mean")
q_values = sorted([decode("q10", expected), decode("q50", expected), decode("q90", expected)])
probability_up = _sigmoid(float(values.get("logit_up", 0.0)))
probability_up = _sigmoid(
float(values.get("logit_tp_first", values.get("logit_up", 0.0)))
)
cap = _prediction_cap(closes, horizon, round_trip_cost)
if str(entry.get("target_transform", "")) == "barrier_net_return":
stop_percent = _clamp(_float_entry(entry, "target_stop_loss_percent", 0.04), 0.003, 0.08)
take_percent = _clamp(_float_entry(entry, "target_take_profit_percent", 0.035), 0.003, 0.20)
cap = max(
cap,
abs(math.log(1.0 - stop_percent) - round_trip_cost),
abs(math.log(1.0 + take_percent) - round_trip_cost),
)
expected = _clamp(expected, -cap, cap)
q10 = _clamp(q_values[0], -cap, cap)
q50 = _clamp(q_values[1], -cap, cap)
@@ -1267,6 +1333,11 @@ def _decode_multi_horizon_prediction(
"q50": q50,
"q90": q90,
"probability_up": probability_up,
"probability_take_profit_first": (
probability_up
if str(entry.get("target_transform", "")) == "barrier_net_return"
else None
),
"volatility_scale": vol_scale,
"validation_mae": mae,
"baseline_mae": base_mae,
@@ -1697,6 +1768,10 @@ def _public_horizon_forecasts(prediction: dict[str, Any]) -> dict[str, Any]:
"quantile_50_percent": round((math.exp(float(row.get("q50", 0.0))) - 1) * 100, 4),
"quantile_90_percent": round((math.exp(float(row.get("q90", 0.0))) - 1) * 100, 4),
}
if isinstance(row.get("probability_take_profit_first"), (int, float)):
public[key]["probability_take_profit_first"] = round(
_clamp(float(row["probability_take_profit_first"]), 0.0, 1.0), 4
)
return public
@@ -1728,6 +1803,10 @@ def _dot(left: list[float], right: list[float]) -> float:
return sum(left[index] * right[index] for index in range(min(len(left), len(right))))
def _gelu(value: float) -> float:
return 0.5 * value * (1.0 + math.erf(value / math.sqrt(2.0)))
def _return_scale(returns: list[float]) -> float:
recent = returns[-120:] if len(returns) > 120 else returns
values = sorted(abs(value) for value in recent if math.isfinite(value))
@@ -1772,6 +1851,41 @@ def _prediction_cap(closes: list[float], horizon: int, round_trip_cost: float) -
return max(base * 1.5 + round_trip_cost, 0.0005)
def _barrier_outcome(
candles: list[Candle],
*,
end_index: int,
horizon: int,
stop_loss_percent: float,
take_profit_percent: float,
round_trip_cost: float,
) -> tuple[float, float] | None:
"""Return after-cost log PnL and whether TP was reached before SL."""
entry_index = end_index + 1
exit_index = end_index + max(1, horizon)
if entry_index >= len(candles) or exit_index >= len(candles):
return None
entry = float(candles[entry_index].open)
if entry <= 0:
return None
stop = entry * (1.0 - _clamp(stop_loss_percent, 0.003, 0.08))
take = entry * (1.0 + _clamp(take_profit_percent, 0.003, 0.20))
for index in range(entry_index, exit_index + 1):
candle = candles[index]
stop_hit = float(candle.low) <= stop
take_hit = float(candle.high) >= take
# OHLC data cannot reveal intrabar ordering, so ties are resolved
# conservatively as stop-loss first.
if stop_hit:
return math.log(stop / entry) - round_trip_cost, 0.0
if take_hit:
return math.log(take / entry) - round_trip_cost, 1.0
terminal = float(candles[exit_index].close)
if terminal <= 0:
return None
return math.log(terminal / entry) - round_trip_cost, 0.0
def _sigmoid(value: float) -> float:
if value >= 40:
return 1.0
@@ -1813,6 +1927,21 @@ def _reason(
return f"model {model}: forecast {expected_return_percent:.3f}%, P(up)={probability_up:.2f}, skill={skill:.3f}"
def _barrier_reason(
model: str,
expected_return_percent: float,
probability_take_profit_first: float,
skill: float,
block_entry: bool,
) -> str:
state = "entry blocked" if block_entry else "entry evaluated"
return (
f"model {model}: expected net {expected_return_percent:.3f}%, "
f"P(TP before SL)={probability_take_profit_first:.2f}, "
f"skill={skill:.3f}; {state}"
)
def _normal_cdf(value: float) -> float:
return 0.5 * (1 + math.erf(value / math.sqrt(2)))