82 lines
3.0 KiB
Python
82 lines
3.0 KiB
Python
"""Analyze price deviation from SMA(24) on 1h candles — to calibrate lock threshold."""
|
|
import sys
|
|
sys.path.insert(0, '/root/grid-bot')
|
|
|
|
from api import TradernetAPI
|
|
from config import (
|
|
TRADERNET_PUBLIC_KEY, TRADERNET_PRIVATE_KEY, SYMBOL,
|
|
TRADERNET_BASE_URL, TRADERNET_LOGIN, TRADERNET_PASSWORD,
|
|
)
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
api = TradernetAPI(
|
|
TRADERNET_PUBLIC_KEY, TRADERNET_PRIVATE_KEY,
|
|
TRADERNET_LOGIN, TRADERNET_PASSWORD, TRADERNET_BASE_URL,
|
|
)
|
|
|
|
# Fetch 7 days of 1h candles
|
|
dt_to = datetime.now(tz=timezone.utc)
|
|
dt_from = dt_to - timedelta(days=7)
|
|
raw = api.get_hloc_sync(
|
|
SYMBOL, 60,
|
|
dt_from.strftime("%d.%m.%Y %H:%M"),
|
|
dt_to.strftime("%d.%m.%Y %H:%M"),
|
|
0,
|
|
)
|
|
hloc = raw.get("hloc", {}).get(SYMBOL, [])
|
|
ts_lst = raw.get("xSeries", {}).get(SYMBOL, [])
|
|
|
|
candles = [
|
|
{"t": int(ts_lst[i]), "c": float(h[3])}
|
|
for i, h in enumerate(hloc[:len(ts_lst)])
|
|
]
|
|
closes = [c["c"] for c in candles]
|
|
print(f"Got {len(closes)} hourly closes: {closes[0]:.0f} → {closes[-1]:.0f}\n")
|
|
|
|
# Compute rolling SMA(24) and deviation in %
|
|
sma_p = 24
|
|
devs = [] # (timestamp, price, sma, dev_pct, atr_1h)
|
|
atr_p = 14
|
|
|
|
# ATR first
|
|
atr_candles = [{"h": float(h[1]), "l": float(h[2]), "c": float(h[3])} for h in hloc[:len(ts_lst)]]
|
|
trs = []
|
|
for i in range(1, len(atr_candles)):
|
|
h, l, pc = atr_candles[i]["h"], atr_candles[i]["l"], atr_candles[i-1]["c"]
|
|
tr = max(h-l, abs(h-pc), abs(l-pc))
|
|
trs.append(tr)
|
|
atr_14 = sum(trs[-atr_p:]) / atr_p if len(trs) >= atr_p else None
|
|
atr_dollar = atr_14
|
|
print(f"ATR(14) on 1h = ${atr_dollar:.2f}\n")
|
|
|
|
for i in range(sma_p, len(closes)):
|
|
sma = sum(closes[i-sma_p:i]) / sma_p
|
|
price = closes[i]
|
|
ts = candles[i]["t"]
|
|
dev_pct = (price - sma) / sma * 100
|
|
devs.append((ts, price, sma, dev_pct))
|
|
|
|
print(f"{'Time (UTC)':<22} {'Price':>10} {'SMA(24)':>10} {'Dev%':>7} {'Dev$':>8} ATR-based")
|
|
print("-" * 75)
|
|
for ts, price, sma, dev_pct in devs[-48:]: # last 48h
|
|
dt = datetime.fromtimestamp(ts, tz=timezone.utc).strftime("%Y-%m-%d %H:%M")
|
|
dev_dollar = price - sma
|
|
dev_atr = abs(dev_dollar) / atr_dollar if atr_dollar else 0
|
|
lock_1p5 = "🔒" if abs(dev_pct) > 1.5 else " "
|
|
lock_2p0 = "🔒" if abs(dev_pct) > 2.0 else " "
|
|
lock_3p0 = "🔒" if abs(dev_pct) > 3.0 else " "
|
|
print(f"{dt:<22} {price:>10,.0f} {sma:>10,.0f} {dev_pct:>+6.2f}% {dev_dollar:>+8,.0f} "
|
|
f"@1.5%{lock_1p5} @2.0%{lock_2p0} @3.0%{lock_3p0}")
|
|
|
|
print("\n=== Deviation distribution ===")
|
|
pcts = [d[3] for d in devs]
|
|
abs_pcts = [abs(p) for p in pcts]
|
|
import statistics
|
|
print(f" Mean abs dev: {statistics.mean(abs_pcts):.2f}%")
|
|
print(f" Median abs dev: {statistics.median(abs_pcts):.2f}%")
|
|
print(f" Max abs dev: {max(abs_pcts):.2f}%")
|
|
print(f" % time |dev| < 1%: {sum(1 for p in abs_pcts if p < 1.0)/len(abs_pcts)*100:.1f}%")
|
|
print(f" % time |dev| < 2%: {sum(1 for p in abs_pcts if p < 2.0)/len(abs_pcts)*100:.1f}%")
|
|
print(f" % time |dev| < 3%: {sum(1 for p in abs_pcts if p < 3.0)/len(abs_pcts)*100:.1f}%")
|
|
print(f" % time |dev| > 3%: {sum(1 for p in abs_pcts if p > 3.0)/len(abs_pcts)*100:.1f}%")
|