"""Moteur de signal en direct, en paper trading.
Applique la strategie publiee par `analyze.py --save` (strategy.json) aux donnees que collector.py et
external.py ecrivent en continu. Deux types de strategie :
- composite : l'indicateur hybride (gradient boosting) donne pour chaque crypto P(hausse de x % d'abord),
P(baisse de x % d'abord) et l'esperance de gain ; signal si elle depasse la marge validee ;
- ridge : score lineaire et seuils (ancien modele, garde comme reference).
Chaque signal = une mise fictive de unit_usd $ (achat, ou vente a decouvert de meme valeur) avec objectif, stop
et duree max ; plusieurs positions peuvent courir en meme temps (cooldown par actif, plafond de positions).
Les resultats vont dans la table `signals`, qu'analyze.py affiche a chaque reentrainement : c'est la boucle
observation -> adaptation. L'etat de l'indicateur est affiche en direct dans indicateur.html.
Coupe-circuit : si les --kill-after derniers signaux d'une strategie sont perdants au total, le moteur se met
en pause jusqu'a la publication d'une nouvelle strategie.
AUCUN ordre reel n'est passe. Brancher un courtier dans on_signal() seulement apres des semaines de
resultats paper positifs.
Usage :
python signal_engine.py
python signal_engine.py --every 30 --kill-after 20 --assets btc,eth
"""
import argparse
import html
import json
import math
import os
import sys
import time
from analyze import HISTORY, Ext, Series, ensure_klines, exit_levels, features_at, load_polymarket, usd_pnl
from technical import RULE_LABELS, RULES, TF_NAMES
from common import SYMBOLS, TF_SECONDS, connect, default_db_path, single_instance
def on_signal(sig):
"""Point d'accroche pour un vrai courtier (non implemente volontairement)."""
side = "ACHAT" if sig["side"] > 0 else "VENTE"
print(f"{time.strftime('%H:%M:%S')} >>> SIGNAL {side} {sig['asset'].upper()} {sig['unit_usd']:g} $ "
f"@ {sig['entry']:.4f} | objectif {sig['target']:.4f} | stop {sig['stop']:.4f} | "
f"echeance {time.strftime('%H:%M', time.localtime(sig['expiry']))} | {sig['why']}", flush=True)
def load_strategy(path, cache):
try:
mtime = os.path.getmtime(path)
except OSError:
return None
if cache.get("mtime") != mtime:
with open(path, encoding="utf-8") as f:
cache["s"] = json.load(f)
cache["mtime"] = mtime
s = cache["s"]
state = "ACTIVE" if s.get("active") else "inactive"
print(f"{time.strftime('%H:%M:%S')} strategie chargee ({time.strftime('%Y-%m-%d %H:%M', time.localtime(s['created']))}) "
f"-> {state} : {s.get('reason')}", flush=True)
if s.get("active"):
v = s["validation"]
print(f" {s.get('type', 'ridge')} | horizon {s['horizon']} | "
f"{'achat + vente' if s['allow_short'] else 'achat seul'} | sortie {s['exit']} | "
f"validation : {v['n']} trades, {v['usd']:+.2f} $, PF {v['pf']:.2f}", flush=True)
return cache["s"]
def evaluate(strat, r, model_dir, seen):
"""Decision pour une crypto a l'instant present : {'side', 'score', 'why', 'info'}.
seen : declencheurs classiques deja traites (un declencheur ne donne lieu qu'a une decision)."""
if strat.get("type") == "meta":
import numpy as np
import composite as cp
model = cp.load(os.path.join(model_dir, strat["model_file"]))
feats, best = strat["features"], None
for rule, tf, d, end in r.get("_events") or ():
k = (r["asset"], rule, tf, end)
if k in seen or (d < 0 and not strat["allow_short"]):
continue
seen[k] = end
m = dict(r, m_rule=RULES.index(rule), m_tf=TF_NAMES.index(tf), m_dir=d)
vec = np.array([[np.nan if m.get(f) is None else m[f] for f in feats]], dtype=float)
p = float(cp.probs(model, vec)[0][0])
if best is None or p > best[0]:
best = (p, rule, tf, d, vec)
if best is None:
return {"side": 0, "score": 0.0, "why": "aucun nouveau declencheur classique", "info": {}}
p, rule, tf, d, vec = best
side = d if p >= strat["threshold"] else 0
# seuls les facteurs qui pesent vraiment (>= 0,1 point de chance de reussite)
factors = [f for f in cp.attribution(model, vec, feats, strat["medians"], lambda up, dn: up * 100)
if abs(f[2]) >= 0.1][:3]
info = {"rule": f"{RULE_LABELS[rule]} {tf}", "dir": d, "p_success": p, "threshold": strat["threshold"],
"factors": [{"name": cp.label(f), "value": v, "contrib": c} for f, v, c in factors]}
why = (f"{info['rule']} ({'achat' if d > 0 else 'vente'}) : reussite estimee {p:.0%} "
f"(seuil {strat['threshold']:.0%}) | " +
(", ".join(f"{x['name']} {x['contrib']:+.1f} pts" for x in info["factors"])
or "aucun facteur ne se demarque"))
return {"side": side, "score": p, "why": why, "info": info}
if strat.get("type") == "composite":
import numpy as np
import composite as cp
model = cp.load(os.path.join(model_dir, strat["model_file"]))
feats = strat["features"]
vec = np.array([[np.nan if r.get(f) is None else r[f] for f in feats]], dtype=float)
x_bps, cost = strat["move"] * 100, strat["cost_bps"]
up, dn = cp.probs(model, vec)
evl, evs = cp.expectancy(up, dn, x_bps, strat["neutral"], cost)
side = int(cp.decide(evl, evs, strat["margin"], strat["allow_short"])[0])
lean = side or (1 if evl[0] >= evs[0] else -1)
# seuls les facteurs qui pesent vraiment (>= 0,1 pb d'esperance)
factors = [f for f in cp.attribution(model, vec, feats, strat["medians"],
lambda u, d: cp.expectancy(u, d, x_bps, strat["neutral"], cost)
[0 if lean > 0 else 1]) if abs(f[2]) >= 0.1][:3]
info = {"p_up": float(up[0]), "p_dn": float(dn[0]), "ev_long": float(evl[0]), "ev_short": float(evs[0]),
"factors": [{"name": cp.label(f), "value": v, "contrib": c} for f, v, c in factors]}
why = (f"P(+{strat['move']:g}%) {info['p_up']:.0%} / P(-{strat['move']:g}%) {info['p_dn']:.0%} | esperance "
f"{max(info['ev_long'], info['ev_short']):+.1f} pb | " +
(", ".join(f"{x['name']} {x['contrib']:+.1f}" for x in info["factors"])
or "aucun facteur ne se demarque"))
return {"side": side, "score": float(evl[0] if lean > 0 else evs[0]), "why": why, "info": info}
z = strat["ym"] + sum(f["w"] * (r[f["name"]] - f["mean"]) / f["sd"] for f in strat["features"]
if r.get(f["name"]) is not None)
side = 1 if z >= strat["hi"] else (-1 if strat["allow_short"] and z <= strat["lo"] else 0)
return {"side": side, "score": z, "why": f"score {z:+.2f}", "info": {"score": z}}
def close_due(con, series, now, cost_bps):
"""Ferme les positions dont l'objectif, le stop ou l'echeance est atteint (bougies 1 min cloturees)."""
closed = 0
for sid, asset, ts, side, entry, target, stop, expiry, unit in con.execute(
"""SELECT id, asset, ts, side, entry, target, stop, expiry, unit_usd
FROM signals WHERE status = 'open'""").fetchall():
sp = series.get(asset)
if sp is None:
continue
exit_px = why = None
m = (int(ts) + 59) // 60 * 60
while m + 60 <= min(now, expiry) and target is not None:
i = sp.idx.get(m)
if i is not None:
hi, lo = sp.h[i], sp.l[i]
if (side > 0 and lo <= stop) or (side < 0 and hi >= stop):
exit_px, why, t_exit = stop, "stop", m + 60
break
if (side > 0 and hi >= target) or (side < 0 and lo <= target):
exit_px, why, t_exit = target, "objectif", m + 60
break
m += 60
if exit_px is None and now >= expiry:
exit_px, why, t_exit = sp.price(expiry), "duree", expiry
if exit_px is None:
continue
g = side * math.log(exit_px / entry)
pnl = usd_pnl(g, side, unit, cost_bps)
con.execute("""UPDATE signals SET status = 'closed', exit_ts = ?, exit_price = ?, exit_reason = ?, pnl_usd = ?
WHERE id = ?""", (t_exit, exit_px, why, pnl, sid))
print(f"{time.strftime('%H:%M:%S')} <<< ferme #{sid} {asset.upper()} {'ACHAT' if side > 0 else 'VENTE'} "
f"({why}) {pnl:+.3f} $", flush=True)
closed += 1
con.commit()
return closed
def killed(con, strat, n):
rows = con.execute("""SELECT pnl_usd FROM signals WHERE strategy_id = ? AND status = 'closed'
ORDER BY exit_ts DESC LIMIT ?""", (strat["created"], n)).fetchall()
return len(rows) >= n and sum(r[0] for r in rows) < 0
# ---------------------------------------------------------------- affichage en direct
PAGE = """
Indicateur composite
Indicateur composite
__SUB__
__CARDS__
"""
def fmt_value(v):
return "-" if v is None else f"{v:.3g}"
def write_live(path, strat, states):
if strat and strat.get("type") == "meta" and strat.get("active"):
sub = (f"Signaux classiques filtres par l'algo · horizon {strat['horizon']} · objectif +{strat['move']:g} % / "
f"stop -{strat['move']:g} % · mis a jour {time.strftime('%H:%M:%S')}")
elif strat and strat.get("type") == "composite" and strat.get("active"):
sub = (f"Horizon {strat['horizon']} · objectif +{strat['move']:g} % / stop -{strat['move']:g} % · signal si "
f"esperance > {strat['margin']} pb apres frais · mis a jour {time.strftime('%H:%M:%S')}")
else:
sub = (f"Aucune strategie composite active ({html.escape((strat or {}).get('reason', 'pas encore publiee'))}). "
f"Mis a jour {time.strftime('%H:%M:%S')}")
cards = []
for a, st in states.items():
info = st.get("info") or {}
if "p_success" in info:
tag = ('ACHAT' if st["side"] > 0 else
'VENTE' if st["side"] < 0 else 'ecarte')
facts = "".join(f"
')
tmp = path + ".tmp"
with open(tmp, "w", encoding="utf-8") as f:
f.write(PAGE.replace("__SUB__", sub).replace("__CARDS__", "".join(cards)))
os.replace(tmp, path)
def main():
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("--db", default=default_db_path())
ap.add_argument("--strategy", help="chemin de strategy.json (defaut : a cote de la base)")
ap.add_argument("--assets", default="btc,eth,sol,xrp")
ap.add_argument("--every", type=int, default=30, help="periode d'evaluation (s)")
ap.add_argument("--kill-after", type=int, default=20, help="pause si les N derniers signaux perdent au total")
args = ap.parse_args()
sys.stdout.reconfigure(errors="replace")
data_dir = os.path.dirname(os.path.abspath(args.db))
path = args.strategy or os.path.join(data_dir, "strategy.json")
live_path = os.path.join(data_dir, "indicateur.html")
assets = [a.strip() for a in args.assets.split(",") if a.strip()]
lock = single_instance("signal_engine", args.db) # noqa: F841 (garde le verrou ouvert)
con = connect(args.db)
cache, paused_for, last_status, seen = {}, None, 0, {}
print(f"moteur de signal (paper) | base {args.db} | strategie {path}\naffichage en direct : {live_path}", flush=True)
while True:
loop_t = time.time()
states = {}
try:
strat = load_strategy(path, cache)
tfs = (strat or {}).get("tfs", ["5m", "15m", "1h", "4h"])
now = int(time.time())
since = now - max(TF_SECONDS[t] for t in tfs) - 4 * 3600
open_assets = {r[0] for r in con.execute("SELECT DISTINCT asset FROM signals WHERE status = 'open'")}
active = bool(strat and strat.get("active"))
series = {}
for a in assets:
if active or a in open_assets:
ensure_klines(con, SYMBOLS[a], now - HISTORY, now)
series[a] = Series(con, SYMBOLS[a], now - HISTORY)
close_due(con, series, now, (strat or {}).get("cost_bps", 20.0))
btc_ref = series.get("btc")
if active and btc_ref is None:
ensure_klines(con, "BTCUSDT", now - HISTORY, now)
btc_ref = Series(con, "BTCUSDT", now - HISTORY)
for k in [k for k, end in seen.items() if end < now - 86400]:
del seen[k]
paused = active and paused_for == strat["created"]
if active and not paused and killed(con, strat, args.kill_after):
paused_for, paused = strat["created"], True
print(f"{time.strftime('%H:%M:%S')} !!! coupe-circuit : les {args.kill_after} derniers signaux perdent "
f"au total -> pause jusqu'a la prochaine strategie", flush=True)
if active:
layers = load_polymarket(con, tfs, since)
ext = Ext(con, assets, now - HISTORY)
n_open = con.execute("SELECT COUNT(*) FROM signals WHERE status = 'open'").fetchone()[0]
for a in assets:
r = features_at(a, now, layers, series[a], tfs, ext, btc_ref, 2 * args.every)
if r is None:
states[a] = {"msg": "donnees en direct incompletes (collecteur arrete ?)"}
continue
if strat.get("lab"): # indicateurs inventes par le laboratoire et utilises par la strategie
import lab
for spec in strat["lab"]:
r[f"lab_{spec['id']}"] = lab.value_row(spec, r)
ev = evaluate(strat, r, data_dir, seen)
states[a] = ev
side = ev["side"]
if paused or not side or n_open >= strat["max_open"]:
continue
last = con.execute("SELECT MAX(ts) FROM signals WHERE asset = ? AND strategy_id = ?",
(a, strat["created"])).fetchone()[0]
if last and now - last < strat["cooldown"]:
continue
S = r["_S"]
target, stop = exit_levels(S, r["_rv"], side, strat["X"], strat["exit"])
sig = {"strategy_id": strat["created"], "asset": a, "ts": now, "side": side, "score": ev["score"],
"entry": S, "target": target, "stop": stop, "expiry": now + strat["X"],
"unit_usd": strat["unit_usd"], "status": "open", "info": json.dumps(ev["info"]),
"why": ev["why"]}
con.execute("""INSERT INTO signals(strategy_id, asset, ts, side, score, entry, target, stop, expiry,
unit_usd, status, info) VALUES(:strategy_id, :asset, :ts, :side, :score, :entry,
:target, :stop, :expiry, :unit_usd, :status, :info)""", sig)
con.commit()
n_open += 1
on_signal(sig)
write_live(live_path, strat, states)
if loop_t - last_status >= 300:
n_open, n_closed, tot = con.execute(
"""SELECT SUM(status = 'open'), SUM(status = 'closed'), COALESCE(SUM(pnl_usd), 0)
FROM signals""").fetchone()
print(f"{time.strftime('%H:%M:%S')} etat : {n_open or 0} positions ouvertes | {n_closed or 0} fermees | "
f"resultat cumule {tot:+.2f} $ | strategie {'active' if active else 'inactive'}"
f"{' (en pause)' if paused else ''}", flush=True)
for a, st in states.items():
print(f" {a.upper()} : {st.get('why') or st.get('msg')}", flush=True)
last_status = loop_t
except KeyboardInterrupt:
raise
except Exception as e:
print(f"{time.strftime('%H:%M:%S')} erreur : {e}", flush=True)
time.sleep(max(1.0, args.every - (time.time() - loop_t)))
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
print("arret")