make xauusd atr breakout edge deployable

This commit is contained in:
root
2026-04-06 13:29:09 +00:00
parent 27382576ea
commit efb814c0e3
3 changed files with 81 additions and 34 deletions
+1
View File
@@ -8,5 +8,6 @@ The default gate is intentionally responsive so PPE produces more events than a
- `breakout_window`: `12`
- `atr_mult`: `0.05`
- no runtime package dependencies
If it overtrades, raise `atr_mult` first. If it is still too quiet, shorten `breakout_window`.
@@ -10,14 +10,8 @@
"breakout_window": 12,
"atr_mult": 0.05
},
"training_requirements": [
"numpy",
"pandas"
],
"inference_requirements": [
"numpy",
"pandas"
],
"training_requirements": [],
"inference_requirements": [],
"symbol": "XAUUSD",
"description": "XAUUSD ATR breakout rule baseline",
"disclaimer": "Educational template only. Not financial advice."
+78 -26
View File
@@ -1,23 +1,47 @@
import numpy as np
import pandas as pd
def _clean_candles(candles):
cleaned = []
for candle in candles:
try:
cleaned.append({
"open": float(candle[0]),
"high": float(candle[1]),
"low": float(candle[2]),
"close": float(candle[3]),
"volume": float(candle[4]) if len(candle) > 4 else 1.0,
})
except Exception:
continue
return cleaned
def _df(candles):
df = pd.DataFrame(candles, columns=["open", "high", "low", "close", "volume"])
for col in df.columns:
df[col] = pd.to_numeric(df[col], errors="coerce")
return df.dropna().reset_index(drop=True)
def _atr(df, n):
prev = df["close"].shift()
tr = pd.concat([(df["high"] - df["low"]), (df["high"] - prev).abs(), (df["low"] - prev).abs()], axis=1).max(axis=1)
return tr.ewm(span=n, adjust=False).mean()
def _atr(candles, n):
if len(candles) < 2:
return 0.0
alpha = 2.0 / (n + 1.0)
value = None
prev_close = candles[0]["close"]
for candle in candles[1:]:
tr = max(
candle["high"] - candle["low"],
abs(candle["high"] - prev_close),
abs(candle["low"] - prev_close),
)
value = tr if value is None else value + alpha * (tr - value)
prev_close = candle["close"]
return float(value or 0.0)
def train(data, config):
params = config.get("parameters", {})
return {"params": params, "name": "xauusd_atr_breakout"}, {"training_bars": int(len(data)), "model": "rule_baseline"}
return {
"params": {
"lookback": int(params.get("lookback", 64)),
"atr_window": int(params.get("atr_window", 14)),
"breakout_window": int(params.get("breakout_window", 12)),
"atr_mult": float(params.get("atr_mult", 0.05)),
},
"name": "xauusd_atr_breakout",
}, {"training_bars": int(len(data)), "model": "edge_rule_baseline"}
def predict(model, market_data, config):
@@ -26,16 +50,44 @@ def predict(model, market_data, config):
atr_window = int(params.get("atr_window", 14))
breakout_window = int(params.get("breakout_window", 12))
atr_mult = float(params.get("atr_mult", 0.05))
candles = market_data.get("candles", [])
candles = _clean_candles(market_data.get("candles", []))
if len(candles) < lookback:
return {"signal": "HOLD", "confidence": 0.0, "metadata": {"reason": "not_enough_candles"}}
df = _df(candles[-lookback:])
atr = float(_atr(df, atr_window).iloc[-1])
close = float(df["close"].iloc[-1])
high = float(df["high"].iloc[-breakout_window:-1].max())
low = float(df["low"].iloc[-breakout_window:-1].min())
if close > high + atr * atr_mult:
return {"signal": "UP", "confidence": 0.64, "metadata": {"breakout": "high", "atr": atr}}
if close < low - atr * atr_mult:
return {"signal": "DOWN", "confidence": 0.64, "metadata": {"breakout": "low", "atr": atr}}
return {"signal": "HOLD", "confidence": 0.2, "metadata": {"high": high, "low": low, "atr": atr}}
return {
"signal": "HOLD",
"confidence": 0.0,
"metadata": {"reason": "not_enough_candles", "got": len(candles), "need": lookback},
}
window = candles[-lookback:]
previous = window[-breakout_window - 1:-1]
if len(previous) < breakout_window:
return {"signal": "HOLD", "confidence": 0.0, "metadata": {"reason": "not_enough_breakout_window"}}
atr = _atr(window, atr_window)
close = window[-1]["close"]
high = max(candle["high"] for candle in previous)
low = min(candle["low"] for candle in previous)
upper = high + atr * atr_mult
lower = low - atr * atr_mult
if close > upper:
edge = (close - upper) / max(atr, 1e-9)
return {
"signal": "UP",
"confidence": round(min(0.55 + edge, 0.92), 4),
"metadata": {"breakout": "high", "atr": round(atr, 6), "upper": round(upper, 6), "close": round(close, 6)},
}
if close < lower:
edge = (lower - close) / max(atr, 1e-9)
return {
"signal": "DOWN",
"confidence": round(min(0.55 + edge, 0.92), 4),
"metadata": {"breakout": "low", "atr": round(atr, 6), "lower": round(lower, 6), "close": round(close, 6)},
}
return {
"signal": "HOLD",
"confidence": 0.2,
"metadata": {"high": round(high, 6), "low": round(low, 6), "atr": round(atr, 6), "upper": round(upper, 6), "lower": round(lower, 6)},
}