95 lines
4.2 KiB
Python
95 lines
4.2 KiB
Python
from __future__ import annotations
|
|
|
|
import math
|
|
import os
|
|
from typing import Any
|
|
|
|
from crypto_spot_bot.storage import Storage
|
|
|
|
|
|
def shadow_gate_snapshot(storage: Storage, model_sha256: str) -> dict[str, Any]:
|
|
minimum_settled = _int_env("SHADOW_GATE_MIN_SETTLED", 300)
|
|
minimum_eligible = _int_env("SHADOW_GATE_MIN_ELIGIBLE", 30)
|
|
minimum_symbols = _int_env("SHADOW_GATE_MIN_SYMBOLS", 2)
|
|
minimum_profit_factor = _float_env("SHADOW_GATE_MIN_PROFIT_FACTOR", 1.10)
|
|
minimum_direction_accuracy = _float_env("SHADOW_GATE_MIN_DIRECTION_ACCURACY", 0.52)
|
|
maximum_brier = _float_env("SHADOW_GATE_MAX_BRIER", 0.25)
|
|
rows = storage.shadow_prediction_rows(model_sha256=model_sha256) if model_sha256 else []
|
|
settled = [row for row in rows if row.get("settled_at")]
|
|
eligible = [row for row in settled if bool(row.get("eligible_signal"))]
|
|
eligible_returns = [float(row.get("actual_return_percent", 0.0) or 0.0) for row in eligible]
|
|
gross_profit = sum(max(0.0, value) for value in eligible_returns)
|
|
gross_loss = abs(sum(min(0.0, value) for value in eligible_returns))
|
|
profit_factor = gross_profit / gross_loss if gross_loss > 1e-12 else (float("inf") if gross_profit > 0 else 0.0)
|
|
correct = sum(
|
|
1
|
|
for row in settled
|
|
if (float(row.get("expected_return_percent", 0.0) or 0.0) >= 0)
|
|
== (float(row.get("actual_return_percent", 0.0) or 0.0) >= 0)
|
|
)
|
|
direction_accuracy = correct / len(settled) if settled else 0.0
|
|
brier_values = [
|
|
(
|
|
max(0.0, min(1.0, float(row.get("probability_up", 0.5) or 0.5)))
|
|
- float(int(row.get("take_profit_first", 0) or 0))
|
|
)
|
|
** 2
|
|
for row in settled
|
|
if row.get("take_profit_first") is not None
|
|
]
|
|
brier = sum(brier_values) / len(brier_values) if brier_values else 1.0
|
|
symbols = sorted({str(row.get("symbol") or "") for row in eligible if row.get("symbol")})
|
|
checks = {
|
|
"minimum_settled": len(settled) >= minimum_settled,
|
|
"minimum_eligible": len(eligible) >= minimum_eligible,
|
|
"minimum_symbols": len(symbols) >= minimum_symbols,
|
|
"positive_average_net": bool(eligible_returns) and sum(eligible_returns) / len(eligible_returns) > 0.0,
|
|
"profit_factor": profit_factor >= minimum_profit_factor,
|
|
"direction_accuracy": direction_accuracy >= minimum_direction_accuracy,
|
|
"brier": brier <= maximum_brier,
|
|
}
|
|
enough_data = checks["minimum_settled"] and checks["minimum_eligible"] and checks["minimum_symbols"]
|
|
passed = enough_data and all(checks.values())
|
|
state = "passed" if passed else ("failed" if enough_data else "collecting")
|
|
return {
|
|
"available": bool(model_sha256),
|
|
"model_sha256": model_sha256,
|
|
"state": state,
|
|
"passed": passed,
|
|
"active_model_unchanged": True,
|
|
"total_predictions": len(rows),
|
|
"pending_predictions": len(rows) - len(settled),
|
|
"settled_predictions": len(settled),
|
|
"eligible_predictions": len(eligible),
|
|
"eligible_symbols": symbols,
|
|
"average_net_percent": round(sum(eligible_returns) / len(eligible_returns), 6) if eligible_returns else 0.0,
|
|
"total_net_percent": round(sum(eligible_returns), 6),
|
|
"win_rate": round(sum(value > 0 for value in eligible_returns) / len(eligible_returns), 6) if eligible_returns else 0.0,
|
|
"profit_factor": round(profit_factor, 6) if math.isfinite(profit_factor) else None,
|
|
"direction_accuracy": round(direction_accuracy, 6),
|
|
"brier": round(brier, 6),
|
|
"criteria": {
|
|
"minimum_settled": minimum_settled,
|
|
"minimum_eligible": minimum_eligible,
|
|
"minimum_symbols": minimum_symbols,
|
|
"minimum_profit_factor": minimum_profit_factor,
|
|
"minimum_direction_accuracy": minimum_direction_accuracy,
|
|
"maximum_brier": maximum_brier,
|
|
},
|
|
"checks": checks,
|
|
}
|
|
|
|
|
|
def _int_env(name: str, default: int) -> int:
|
|
try:
|
|
return max(1, int(os.environ.get(name, str(default))))
|
|
except ValueError:
|
|
return default
|
|
|
|
|
|
def _float_env(name: str, default: float) -> float:
|
|
try:
|
|
return float(os.environ.get(name, str(default)))
|
|
except ValueError:
|
|
return default
|