from __future__ import annotations import argparse import atexit import csv import hashlib import importlib.util import json import math import os import sqlite3 import statistics import time import urllib.parse import urllib.request from datetime import datetime, timedelta, timezone from pathlib import Path def load_feature_module(): try: import build_market_features_v2 as module return module except ModuleNotFoundError: path = Path(__file__).resolve().with_name("build_market_features_v2.py") spec = importlib.util.spec_from_file_location("shadow_preentry_features", path) if spec is None or spec.loader is None: raise RuntimeError(f"Feature engine non disponibile: {path}") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module features = load_feature_module() TF_MINUTES = {"5M": 5, "15M": 15, "30M": 30, "1H": 60, "2H": 120, "3H": 180, "4H": 240} BYBIT_INTERVAL = {"5M": "5", "15M": "15", "30M": "30", "1H": "60", "2H": "120", "3H": "60", "4H": "240"} EXIT_MARKERS = ( "max exit", "exit bars", "opposite signal", "exit short", "exit long", "short exit", "long exit", "trailing short", "trailing long", "trailing exit", "close entry", "strategy close", "end date", ) def now_utc() -> datetime: return datetime.now(timezone.utc) def parse_dt(value: str) -> datetime: parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00")) return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc) def normalize_text(value) -> str: return " ".join(str(value or "").lower().replace("_", " ").replace("-", " ").split()) def normalize_family(value) -> str: compact = normalize_text(value).replace(" ", "") aliases = { "vptr3": "VPTR3", "vptrv3": "VPTR3", "rettangolo": "RETTANGOLO", "rettangolov2": "RETTANGOLO", "rettangolosimple": "RETTANGOLO_SIMPLE", "rettangolotvsimple": "RETTANGOLO_SIMPLE", "arrettangolotv": "RETTANGOLO_SIMPLE", "matrailing": "MA_TRAILING", "rsiswing": "RSI_SWING_BREAKOUT", "rsiswingbreakout": "RSI_SWING_BREAKOUT", "sqw": "SQW", "range": "RANGE_FIB", "rangefib": "RANGE_FIB", "supertrendbosw": "SUPERTREND_BOSW", "adx": "ADX", } return aliases.get(compact, "OTHER") def normalize_tf(value, family: str) -> str: text = str(value or "").upper().replace(" ", "") aliases = {"5": "5M", "15": "15M", "30": "30M", "60": "1H", "120": "2H", "180": "3H", "240": "4H"} if text in aliases: return aliases[text] if text.endswith("MIN"): text = text[:-3] + "M" if text in TF_MINUTES: return text return "5M" if family == "RETTANGOLO" else "UNKNOWN" def event_type(payload: dict) -> str: try: if float(payload.get("qty", 1) or 0) <= 0: return "EXIT" except (TypeError, ValueError): pass if ( normalize_text(payload.get("intent_source")) == "market position transition" and normalize_text(payload.get("market_position")) == "flat" ): return "EXIT" for field in ("comment", "alert_message", "order_id"): text = normalize_text(payload.get(field)) if any(marker in text for marker in EXIT_MARKERS): return "EXIT" return "ENTRY" def safe_payload_fingerprint(payload: dict) -> str: safe = {key: value for key, value in payload.items() if key.lower() not in {"secret", "api_key", "api_secret", "password"}} return hashlib.sha256(json.dumps(safe, sort_keys=True, default=str).encode("utf-8")).hexdigest() def robust_scales(history: list[dict], names: list[str]) -> dict[str, tuple[float, float]]: result = {} for name in names: values = sorted(float(row[name]) for row in history) q1 = values[(len(values) - 1) // 4] q3 = values[((len(values) - 1) * 3) // 4] result[name] = (statistics.median(values), max(q3 - q1, 1e-9)) return result def vector_distance(left: dict, right: dict, model: dict, scales: dict) -> float: parts = [min(abs(float(left[name]) - float(right[name])) / scales[name][1], 4.0) ** 2 for name in model["numeric_features"]] if model.get("cyclic_hour"): delta = abs(int(left["entry_hour_utc"]) - int(right["entry_hour_utc"])) parts.append((min(delta, 24 - delta) / 6.0) ** 2) parts.extend(0.0 if left[name] == right[name] else 1.0 for name in model["categorical_features"]) return math.sqrt(sum(parts) / len(parts)) def score_vector(vector: dict, model: dict) -> dict: history = model["history"] scales = robust_scales(history, model["numeric_features"]) neighbors = sorted(((vector_distance(vector, row, model, scales), row) for row in history), key=lambda item: item[0])[: int(model["neighbors"])] weights = [math.exp(-item[0]) for item in neighbors] weighted_losses = sum(weight * int(item[1]["loss"]) for weight, item in zip(weights, neighbors)) risk = 100 * (weighted_losses + 1.0) / (sum(weights) + 2.0) confidence = min(100.0, 50.0 * min(1.0, len(history) / 40.0) + 50.0 * sum(weights) / len(weights)) calibration = [] for index, row in enumerate(history): peers = history[:index] + history[index + 1:] if peers: calibration.append(min(vector_distance(row, peer, model, scales) for peer in peers)) ordered = sorted(calibration) limit = ordered[round((len(ordered) - 1) * 0.90)] * 1.25 if ordered else math.inf nearest = neighbors[0][0] if neighbors else math.inf losses = sum(int(item[1]["loss"]) for item in neighbors) if nearest > limit: risk_class = "NO_DATA" reason = f"Fuori campione: distanza {nearest:.2f} > limite {limit:.2f}" else: risk_class = "HIGH_RISK" if risk >= float(model["high_risk_threshold"]) else ("WARNING" if risk >= float(model["threshold"]) else "CLEAR") reason = f"{losses}/{len(neighbors)} precedenti simili in perdita; rischio {risk:.1f}" return { "risk_score": round(risk, 2), "risk_class": risk_class, "confidence": round(confidence, 1), "reason": reason, "nearest_losses": losses, "nearest_count": len(neighbors), "nearest_distance": round(nearest, 4), "ood_limit": round(limit, 4), } def fetch_series(symbol: str, tf: str, end: datetime, limit: int = 500) -> list[dict]: source_tf = "1H" if tf == "3H" else tf minutes = TF_MINUTES[source_tf] start = end - timedelta(minutes=minutes * limit * 2) query = urllib.parse.urlencode({ "category": "linear", "symbol": symbol, "interval": BYBIT_INTERVAL[tf], "start": int(start.timestamp() * 1000), "end": int(end.timestamp() * 1000), "limit": min(limit, 1000), }) request = urllib.request.Request("https://api.bybit.com/v5/market/kline?" + query, headers={"User-Agent": "ShadowPreEntry/1.0"}) with urllib.request.urlopen(request, timeout=20) as response: payload = json.loads(response.read().decode("utf-8")) if payload.get("retCode") != 0: raise RuntimeError(f"Bybit public candles: {payload.get('retMsg')}") rows = [{ "symbol": symbol, "tf": source_tf, "time": datetime.fromtimestamp(int(item[0]) / 1000, timezone.utc), "open": float(item[1]), "high": float(item[2]), "low": float(item[3]), "close": float(item[4]), "volume": float(item[5]), } for item in payload.get("result", {}).get("list", [])] rows.sort(key=lambda row: row["time"]) if tf == "3H": buckets = {} for row in rows: key = row["time"].replace(hour=(row["time"].hour // 3) * 3, minute=0, second=0, microsecond=0) buckets.setdefault(key, []).append(row) rows = [{ "symbol": symbol, "tf": "3H", "time": key, "open": group[0]["open"], "high": max(x["high"] for x in group), "low": min(x["low"] for x in group), "close": group[-1]["close"], "volume": sum(x["volume"] for x in group), } for key, group in sorted(buckets.items()) if len(group) == 3] features.add_advanced_features(rows) return rows def last_closed(group: list[dict], at: datetime, minutes: int) -> tuple[int, dict] | tuple[None, None]: for index in range(len(group) - 1, -1, -1): if group[index]["time"] + timedelta(minutes=minutes) <= at: return index, group[index] return None, None def vptr_vector(symbol: str, tf: str, side: str, at: datetime) -> tuple[dict, str]: group = fetch_series(symbol, tf, at) minutes = TF_MINUTES[tf] index, candle = last_closed(group, at, minutes) if candle is None or index is None: raise RuntimeError("Nessuna candela chiusa disponibile") higher, higher_minutes = features.aggregate(group, minutes) features.add_advanced_features(higher) _, htf = last_closed(higher, at, higher_minutes) btc = fetch_series("BTCUSDT", "4H", at) btc_index, btc_candle = last_closed(btc, at, 240) direction = 1 if side == "LONG" else -1 close = candle["close"] row = { "rsi14_dir_score": direction * ((candle.get("rsi14") or 50) - 50), "adx14": candle.get("adx14"), "di_spread_dir": direction * ((candle.get("plus_di14") or 0) - (candle.get("minus_di14") or 0)), "ema20_slope_5_pct": candle.get("ema20_slope_5_pct"), "ema50_slope_5_pct": candle.get("ema50_slope_5_pct"), "close_vs_ema20_dir_pct": direction * (close - candle["ema20"]) / close * 100, "close_vs_ema50_dir_pct": direction * (close - candle["ema50"]) / close * 100, "trend_efficiency_10": candle.get("trend_efficiency_10"), "realized_vol_20_pct": candle.get("realized_vol_20_pct"), "range_pct": candle.get("range_pct"), "volume_zscore_20": candle.get("volume_zscore_20"), "bb_width_percentile_100": candle.get("bb_width_percentile_100"), "atr_percentile_100": candle.get("atr_percentile_100"), "entry_hour_utc": at.hour, "weekday_utc": at.strftime("%A"), } slope = candle.get("ema50_slope_5_pct") or 0 row["market_regime"] = "TREND_UP" if (candle.get("adx14") or 0) >= 20 and slope > 0 else ("TREND_DOWN" if (candle.get("adx14") or 0) >= 20 and slope < 0 else "RANGE") rank = candle.get("atr_percentile_100") row["volatility_regime"] = "HIGH" if rank is not None and rank >= 70 else ("LOW" if rank is not None and rank <= 30 else "NORMAL") row["htf_return_3_dir_pct"] = None row["htf_close_vs_ema20_dir_pct"] = None row["htf_adx14"] = None if htf: hi = higher.index(htf) previous = higher[max(0, hi - 3)]["close"] row["htf_return_3_dir_pct"] = direction * (htf["close"] - previous) / previous * 100 if previous else None row["htf_close_vs_ema20_dir_pct"] = direction * (htf["close"] - htf["ema20"]) / htf["close"] * 100 row["htf_adx14"] = htf.get("adx14") if btc_candle is None or btc_index is None: raise RuntimeError("Contesto BTC non disponibile") for lookback in (3, 10): previous = btc[max(0, btc_index - lookback)]["close"] row[f"btc_return_{lookback}_dir_pct"] = direction * (btc_candle["close"] - previous) / previous * 100 if previous else None row["btc_rsi14_dir_score"] = direction * ((btc_candle.get("rsi14") or 50) - 50) row["btc_adx14"] = btc_candle.get("adx14") row["mtf_alignment"] = int((row.get("close_vs_ema20_dir_pct") or 0) > 0 and (row.get("htf_close_vs_ema20_dir_pct") or 0) > 0) if any(row.get(name) is None for name in ( "adx14", "ema20_slope_5_pct", "ema50_slope_5_pct", "trend_efficiency_10", "realized_vol_20_pct", "volume_zscore_20", "bb_width_percentile_100", "atr_percentile_100", "htf_return_3_dir_pct", "htf_close_vs_ema20_dir_pct", "htf_adx14", "btc_adx14", )): raise RuntimeError("Feature pre-entry insufficienti") return row, candle["time"].isoformat() def rettangolo_vector(symbol: str, tf: str, side: str, at: datetime) -> tuple[dict, str]: group = fetch_series(symbol, tf, at) index, candle = last_closed(group, at, TF_MINUTES[tf]) if candle is None or index is None or index < 10: raise RuntimeError("Feature RETTANGOLO insufficienti") direction = 1 if side == "LONG" else -1 close = candle["close"] r3 = (close - group[index - 3]["close"]) / group[index - 3]["close"] * 100 r10 = (close - group[index - 10]["close"]) / group[index - 10]["close"] * 100 adx, slope, rank = candle.get("adx14") or 0, candle.get("ema50_slope_5_pct") or 0, candle.get("atr_percentile_100") row = { "side": side, "rsi14_dir_score": direction * ((candle.get("rsi14") or 50) - 50), "adx14": candle.get("adx14"), "di_spread_dir": direction * ((candle.get("plus_di14") or 0) - (candle.get("minus_di14") or 0)), "ema20_slope_dir_pct": direction * (candle.get("ema20_slope_5_pct") or 0), "close_vs_ema20_dir_pct": direction * (close - candle["ema20"]) / close * 100, "close_vs_ema50_dir_pct": direction * (close - candle["ema50"]) / close * 100, "trend_efficiency_10": candle.get("trend_efficiency_10"), "realized_vol_20_pct": candle.get("realized_vol_20_pct"), "range_pct": candle.get("range_pct"), "volume_zscore_20": candle.get("volume_zscore_20"), "bb_width_percentile_100": candle.get("bb_width_percentile_100"), "atr_percentile_100": rank, "return_3_dir_pct": direction * r3, "return_10_dir_pct": direction * r10, "weekday_utc": at.strftime("%A"), "market_regime": "TREND_UP" if adx >= 20 and slope > 0 else ("TREND_DOWN" if adx >= 20 and slope < 0 else "RANGE"), "volatility_regime": "HIGH" if rank is not None and rank >= 70 else ("LOW" if rank is not None and rank <= 30 else "NORMAL"), } if any(row.get(name) is None for name in row if name not in {"side", "weekday_utc", "market_regime", "volatility_regime"}): raise RuntimeError("Feature RETTANGOLO insufficienti") return row, candle["time"].isoformat() def source_connect(path: Path) -> sqlite3.Connection: con = sqlite3.connect(path.resolve().as_uri() + "?mode=ro", uri=True, timeout=5) con.execute("PRAGMA query_only=ON") con.row_factory = sqlite3.Row return con def source_table(con: sqlite3.Connection) -> str: names = {row[0] for row in con.execute("SELECT name FROM sqlite_master WHERE type='table'")} for candidate in ("queue", "webhook_queue"): if candidate in names: return candidate raise RuntimeError("Tabella queue non trovata") def output_connect(path: Path) -> sqlite3.Connection: path.parent.mkdir(parents=True, exist_ok=True) con = sqlite3.connect(path, timeout=10) con.row_factory = sqlite3.Row con.executescript(""" CREATE TABLE IF NOT EXISTS state(account TEXT PRIMARY KEY,last_queue_id INTEGER NOT NULL,initialized_at TEXT NOT NULL); CREATE TABLE IF NOT EXISTS evaluations( id INTEGER PRIMARY KEY AUTOINCREMENT,account TEXT NOT NULL,queue_id INTEGER NOT NULL,request_id TEXT NOT NULL, received_at TEXT NOT NULL,evaluated_at TEXT NOT NULL,event_type TEXT NOT NULL,strategy_raw TEXT,family TEXT NOT NULL, symbol TEXT,side TEXT,timeframe TEXT,model_id TEXT,model_version TEXT,threshold REAL,risk_score REAL,risk_class TEXT NOT NULL, confidence REAL,reason TEXT,candle_time TEXT,status TEXT NOT NULL,error TEXT,payload_fingerprint TEXT NOT NULL, automatic_entry INTEGER NOT NULL DEFAULT 0,automatic_blocking INTEGER NOT NULL DEFAULT 0, UNIQUE(account,request_id)); """) con.commit() return con def export_outputs(con: sqlite3.Connection, output_db: Path) -> None: rows = [dict(row) for row in con.execute("SELECT * FROM evaluations ORDER BY id DESC LIMIT 500")] csv_path = output_db.with_suffix(".csv") if rows: with csv_path.open("w", newline="", encoding="utf-8-sig") as handle: writer = csv.DictWriter(handle, fieldnames=list(rows[0])) writer.writeheader(); writer.writerows(rows) counts = {row[0]: row[1] for row in con.execute("SELECT risk_class,COUNT(*) FROM evaluations GROUP BY risk_class")} payload = { "generated_at": now_utc().isoformat(), "block": 9, "mode": "SHADOW_PREENTRY", "evaluations": int(con.execute("SELECT COUNT(*) FROM evaluations").fetchone()[0]), "counts": counts, "rows": rows[:50], "automatic_entry": False, "automatic_blocking": False, } output_db.with_suffix(".json").write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") def evaluate_row(account: str, row: sqlite3.Row, bundle: dict) -> dict: payload = json.loads(row["payload"]) family = normalize_family(payload.get("strategy")) symbol = str(payload.get("symbol") or "").upper().replace(".P", "") side = "LONG" if str(payload.get("side") or "").lower() in {"buy", "long"} else "SHORT" tf = normalize_tf(payload.get("tf"), family) kind = event_type(payload) result = { "account": str(account), "queue_id": int(row["id"]), "request_id": row["request_id"], "received_at": row["received_at"], "evaluated_at": now_utc().isoformat(), "event_type": kind, "strategy_raw": str(payload.get("strategy") or ""), "family": family, "symbol": symbol, "side": side, "timeframe": tf, "model_id": None, "model_version": None, "threshold": None, "risk_score": None, "risk_class": "NO_DATA", "confidence": 0.0, "reason": None, "candle_time": None, "status": "IGNORED_EXIT" if kind == "EXIT" else "PENDING", "error": None, "payload_fingerprint": safe_payload_fingerprint(payload), "automatic_entry": 0, "automatic_blocking": 0, } if kind == "EXIT": result["reason"] = "Evento di uscita: nessuna valutazione pre-entry" return result model = bundle.get("models", {}).get(family) if not model: result.update(risk_class="NO_MODEL", status="NO_MODEL", reason="Nessun modello dedicato validabile per questa famiglia") return result result.update(model_id=model["model_id"], model_version=model["model_version"], threshold=model["threshold"]) if tf not in TF_MINUTES or not symbol: result.update(status="INSUFFICIENT_DATA", risk_class="NO_DATA", reason="Simbolo o timeframe non disponibile") return result try: at = parse_dt(row["received_at"]) vector, candle_time = vptr_vector(symbol, tf, side, at) if family == "VPTR3" else rettangolo_vector(symbol, tf, side, at) score = score_vector(vector, model) result.update(score, candle_time=candle_time, status="EVALUATED") except Exception as exc: result.update(status="EVALUATION_ERROR", risk_class="NO_DATA", reason="Valutazione non disponibile", error=f"{type(exc).__name__}: {exc}") return result def insert_evaluation(con: sqlite3.Connection, item: dict) -> None: columns = list(item) con.execute( f"INSERT OR IGNORE INTO evaluations ({','.join(columns)}) VALUES ({','.join('?' for _ in columns)})", [item[name] for name in columns], ) def cycle(account: str, source_db: Path, output_db: Path, bundle: dict, initialize_only: bool = False) -> dict: source = source_connect(source_db) output = output_connect(output_db) try: table = source_table(source) maximum = int(source.execute(f"SELECT COALESCE(MAX(id),0) FROM {table}").fetchone()[0]) state = output.execute("SELECT last_queue_id FROM state WHERE account=?", (str(account),)).fetchone() if state is None: output.execute("INSERT INTO state VALUES (?,?,?)", (str(account), maximum, now_utc().isoformat())) output.commit(); export_outputs(output, output_db) return {"status": "INITIALIZED", "cursor": maximum, "processed": 0} cursor = int(state[0]) if initialize_only: return {"status": "ALREADY_INITIALIZED", "cursor": cursor, "processed": 0} rows = source.execute( f"SELECT id,request_id,received_at,payload FROM {table} WHERE id>? ORDER BY id", (cursor,) ).fetchall() for row in rows: insert_evaluation(output, evaluate_row(account, row, bundle)) cursor = int(row["id"]) output.execute("UPDATE state SET last_queue_id=? WHERE account=?", (cursor, str(account))) output.commit() export_outputs(output, output_db) return {"status": "OK", "cursor": cursor, "processed": len(rows)} finally: source.close(); output.close() def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--account", required=True) parser.add_argument("--source-db", type=Path, required=True) parser.add_argument("--output-db", type=Path, required=True) parser.add_argument("--model-bundle", type=Path, required=True) parser.add_argument("--poll-seconds", type=float, default=5.0) parser.add_argument("--pid-file", type=Path) parser.add_argument("--initialize-only", action="store_true") parser.add_argument("--once", action="store_true") args = parser.parse_args() if args.pid_file: args.pid_file.parent.mkdir(parents=True, exist_ok=True) if args.pid_file.exists(): try: old_pid = int(args.pid_file.read_text(encoding="utf-8").strip()) os.kill(old_pid, 0) raise SystemExit(f"Shadow pre-entry gia attivo con PID {old_pid}") except (ValueError, OSError): pass args.pid_file.write_text(str(os.getpid()), encoding="utf-8") def cleanup_pid() -> None: try: if args.pid_file.exists() and args.pid_file.read_text(encoding="utf-8").strip() == str(os.getpid()): args.pid_file.unlink() except OSError: pass atexit.register(cleanup_pid) bundle = json.loads(args.model_bundle.read_text(encoding="utf-8")) while True: try: result = cycle(args.account, args.source_db, args.output_db, bundle, args.initialize_only) print(json.dumps({"at": now_utc().isoformat(), **result}), flush=True) except Exception as exc: print(json.dumps({"at": now_utc().isoformat(), "status": "ERROR", "error": f"{type(exc).__name__}: {exc}"}), flush=True) if args.once or args.initialize_only: raise if args.once or args.initialize_only: break time.sleep(max(1.0, args.poll_seconds)) if __name__ == "__main__": main()