agent.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
confluence-agent — multi-stage crypto confluence pipeline + Telegram alerts.

Pipeline (sab 15m / pred_len=24 / top-200 coins, verify 1h):
  Stage 1  CONFLUENCE   POST :7860/api/confluence  (Fusion + EdgeRank + Kronos)
  Stage 2  KRONOS SCAN  POST :7860/api/screener    (top-200 mcap Kronos forecast)
  Stage 3  MERGE        dono se candidate set + har coin ki direction
  Stage 4  FORECAST     /api/forecast 15m + 1h  → direction dono TF pe verify
  Stage 5  EDGE CONFIRM edge_screener.analyze_coin 15m + 1h  → qualified gate
  Stage 6  NOTIFY       Telegram (dedup ke saath; sirf confirmed signals)

Config: config.json (telegram creds + thresholds).
Chalana: python agent.py            (ek full cycle)
Schedule: run_agent.ps1 har 15m (install_task.ps1 se).
"""
import json
import math
import os
import sys
import time
import traceback
import urllib.error
import urllib.request
from datetime import datetime, timezone

DIR = os.path.dirname(os.path.abspath(__file__))
CONFIG_PATH = os.path.join(DIR, "config.json")
STATE_PATH = os.path.join(DIR, "state.json")
LOG_PATH = os.path.join(DIR, "agent.log")


# ───────────────────────── Config / state ─────────────────────────
def load_config():
    with open(CONFIG_PATH, "r", encoding="utf-8") as f:
        return json.load(f)


def load_state():
    if not os.path.exists(STATE_PATH):
        return {"notified": {}}
    try:
        with open(STATE_PATH, "r", encoding="utf-8") as f:
            return json.load(f)
    except Exception:
        return {"notified": {}}


def save_state(state):
    tmp = STATE_PATH + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        json.dump(state, f, indent=2)
    os.replace(tmp, STATE_PATH)


def log(msg):
    line = f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {msg}"
    print(line, flush=True)
    try:
        with open(LOG_PATH, "a", encoding="utf-8") as f:
            f.write(line + "\n")
    except Exception:
        pass


# ───────────────────────── HTTP ─────────────────────────
def _req(url, data=None, method="GET", timeout=60):
    headers = {"User-Agent": "confluence-agent/1.0"}
    body = None
    if data is not None:
        body = json.dumps(data).encode()
        headers["Content-Type"] = "application/json"
        method = "POST"
    req = urllib.request.Request(url, data=body, headers=headers, method=method)
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return r.status, json.loads(r.read().decode())


def get_json(url, timeout=60):
    _, j = _req(url, timeout=timeout)
    return j


def post_json(url, data, timeout=60):
    return _req(url, data=data, timeout=timeout)


# ───────────────────────── Stage 1: Confluence ─────────────────────────
def run_confluence(cfg):
    base = cfg["backend"]
    body = {
        "tf": cfg["tf_fast"], "pred_len": cfg["pred_len"],
        "fusion_count": cfg["coins"], "edge_count": cfg["coins"],
        "run_kronos": True, "relaxed": cfg.get("relaxed", False),
    }
    try:
        status, _ = post_json(base + "/api/confluence", body, timeout=30)
        if status == 409:
            log("Confluence already running — uska wait karta hoon.")
    except urllib.error.HTTPError as e:
        if e.code != 409:
            raise
        log("Confluence already running (409) — wait.")

    deadline = time.time() + cfg["timeouts"]["confluence_s"]
    poll = cfg["timeouts"]["poll_s"]
    last = ""
    while time.time() < deadline:
        snap = get_json(base + "/api/confluence/status", timeout=30)
        st = snap.get("status")
        prog = (f"fusion {snap.get('fusion_done',0)}/{snap.get('fusion_total',0)} "
                f"edge {snap.get('edge_done',0)}/{snap.get('edge_total',0)} "
                f"kronos {snap.get('kronos_done',0)}/{snap.get('kronos_total',0)}")
        if prog != last:
            log(f"  confluence [{st}] {prog}")
            last = prog
        if st == "done":
            res = snap.get("result") or {}
            rows = res.get("rows", [])
            log(f"Confluence done: {len(rows)} rows (counts={res.get('counts')})")
            return rows
        if st == "error":
            log(f"Confluence error: {snap.get('error')}")
            return []
        time.sleep(poll)
    log("Confluence timeout.")
    return []


# ───────────────────────── Stage 2: Kronos screener ─────────────────────────
def run_screener(cfg):
    base = cfg["backend"]
    body = {"tf": cfg["tf_fast"], "count": cfg.get("screener_coins", cfg["coins"]),
            "pred_len": cfg["pred_len"], "lookback": cfg.get("lookback", 400)}
    try:
        status, _ = post_json(base + "/api/screener", body, timeout=30)
        if status == 409:
            log("Screener already running — wait.")
    except urllib.error.HTTPError as e:
        if e.code != 409:
            raise
        log("Screener already running (409) — wait.")

    deadline = time.time() + cfg["timeouts"]["screener_s"]
    poll = cfg["timeouts"]["poll_s"]
    last = ""
    while time.time() < deadline:
        snap = get_json(base + "/api/screener/status", timeout=30)
        st = snap.get("status")
        prog = f"{snap.get('done',0)}/{snap.get('total',0)} {snap.get('current_symbol','')}"
        if prog != last:
            log(f"  screener [{st}] {prog}")
            last = prog
        if st == "done":
            res = snap.get("result") or {}
            rows = res.get("rows", [])
            log(f"Screener done: {len(rows)} rows")
            return rows
        if st == "error":
            log(f"Screener error: {snap.get('error')}")
            return []
        time.sleep(poll)
    log("Screener timeout.")
    return []


# ───────────────────────── Stage 4: Forecast ─────────────────────────
def get_forecast(cfg, symbol, tf):
    base = cfg["backend"]
    body = {"symbol": symbol, "tf": tf, "pred_len": cfg["pred_len"],
            "lookback": cfg.get("lookback", 400)}
    try:
        _, j = post_json(base + "/api/forecast", body, timeout=180)
        if j.get("ok"):
            return j
    except Exception as e:
        log(f"  forecast {symbol} {tf} fail: {e}")
    return None


# ───────────────────────── Stage 5: Edge confirm ─────────────────────────
_es = None


def _edge_module(cfg):
    global _es
    if _es is None:
        sys.path.insert(0, cfg["edge_screener_dir"])
        import edge_screener as es  # noqa: E402
        _es = es
    return _es


def edge_check(cfg, symbol, tf, direction):
    """Return dict for the requested direction's CoinResult, or None."""
    es = _edge_module(cfg)
    sym = symbol if symbol.endswith("USDT") else symbol + "USDT"
    p = es.Params(tf=tf, limit=cfg.get("edge_limit", 1500))
    try:
        results = es.analyze_coin(sym, p)
    except Exception as e:
        log(f"  edge {sym} {tf} fail: {e}")
        return None
    for r in results:
        if r.side == direction:
            return {
                "qualified": bool(r.qualified),
                "side": r.side, "n": r.n, "exp_R": r.exp_R,
                "ci_lo": r.ci_lo, "ci_hi": r.ci_hi,
                "sig_bars_ago": r.sig_bars_ago,
                "entry": r.entry, "stop": r.stop, "target": r.target,
                "error": r.error,
            }
    return None


# ───────────────────────── Pipeline ─────────────────────────
def _sign_dir(x):
    return "long" if x >= 0 else "short"


def build_candidates(cfg, conf_rows, scr_rows):
    """Merge confluence + screener rows into {symbol: candidate}."""
    cands = {}
    for r in conf_rows:
        sym = r["symbol"]
        as_dir = r.get("as_dir") or (r.get("k_dir") or "")
        cands[sym] = {
            "symbol": sym,
            "dir": as_dir or None,
            "conf_tag": r.get("tag"),
            "conf_confluence": r.get("confluence"),
            "conf_k_move": r.get("k_move_pct"),
            "scr_move": None,
            "sources": ["confluence"],
        }
    for r in scr_rows:
        sym = r["symbol"]
        mv = r.get("pred_move_pct")
        if sym in cands:
            cands[sym]["scr_move"] = mv
            cands[sym]["sources"].append("screener")
            if not cands[sym]["dir"] and mv is not None:
                cands[sym]["dir"] = _sign_dir(mv)
        else:
            cands[sym] = {
                "symbol": sym,
                "dir": _sign_dir(mv) if mv is not None else None,
                "conf_tag": None, "conf_confluence": None, "conf_k_move": None,
                "scr_move": mv, "sources": ["screener"],
            }
    return cands


def shortlist(cfg, cands):
    """Limit deep verification to the strongest, direction-consistent candidates."""
    mn = cfg["min_move_pct"]
    picks = []
    for c in cands.values():
        if not c["dir"]:
            continue
        d = c["dir"]
        conf_agree = c["conf_confluence"] == "AGREE"
        conf_both = c["conf_tag"] == "BOTH"
        # both Kronos reads (confluence k_move, screener move) should not contradict dir
        moves = [m for m in (c["conf_k_move"], c["scr_move"]) if m is not None]
        aligned = all(_sign_dir(m) == d and abs(m) >= mn for m in moves) if moves else False
        strength = max([abs(m) for m in moves], default=0.0)
        # qualify for deep check: confluence agreement OR strong aligned Kronos move
        if conf_agree or conf_both or (aligned and strength >= mn):
            c["_strength"] = strength + (2 if conf_both else 0) + (1 if conf_agree else 0)
            picks.append(c)
    picks.sort(key=lambda c: c["_strength"], reverse=True)
    return picks[: cfg["shortlist_max"]]


def evaluate_candidate(cfg, cand):
    """Run all gates; ALWAYS return a breakdown dict (never None).
    Keys: forecast_ok, edge_ok, confirmed, gates_passed, plus per-stage detail.
    Edge stage skip ho jata hai agar forecast fail kare (waqt bachane ke liye)."""
    sym, d = cand["symbol"], cand["dir"]
    mn = cfg["min_move_pct"]
    out = {**cand, "move15": None, "move1h": None,
           "pass15": False, "pass1h": False, "forecast_ok": False,
           "edge15": None, "edge1h": None, "q15": False, "q1h": False,
           "edge_ok": False, "confirmed": False, "gates_passed": 0, "reason": ""}

    # --- Stage 4: forecast verify ---
    move15 = cand["conf_k_move"] if cand["conf_k_move"] is not None else cand["scr_move"]
    if move15 is None:
        f15 = get_forecast(cfg, sym, cfg["tf_fast"])
        move15 = f15["pred_move_pct"] if f15 else None
    f1h = get_forecast(cfg, sym, cfg["tf_slow"])
    move1h = f1h["pred_move_pct"] if f1h else None
    out["move15"], out["move1h"] = move15, move1h

    if move15 is None or move1h is None:
        out["reason"] = "forecast data missing"
        return out
    out["pass15"] = _sign_dir(move15) == d and abs(move15) >= mn
    out["pass1h"] = _sign_dir(move1h) == d and abs(move1h) >= mn
    out["forecast_ok"] = ((out["pass15"] and out["pass1h"])
                          if cfg["require_forecast_both_tf"]
                          else (out["pass15"] or out["pass1h"]))
    out["gates_passed"] = int(out["pass15"]) + int(out["pass1h"])
    if not out["forecast_ok"]:
        out["reason"] = "forecast direction mismatch"
        return out

    # --- Stage 5: edge confirm ---
    e15 = edge_check(cfg, sym, cfg["tf_fast"], d)
    e1h = edge_check(cfg, sym, cfg["tf_slow"], d)
    out["edge15"], out["edge1h"] = e15, e1h
    out["q15"] = bool(e15 and e15["qualified"])
    out["q1h"] = bool(e1h and e1h["qualified"])
    out["edge_ok"] = ((out["q15"] and out["q1h"]) if cfg["require_edge_both_tf"]
                      else (out["q15"] or out["q1h"]))
    out["gates_passed"] += int(out["q15"]) + int(out["q1h"])
    out["confirmed"] = out["forecast_ok"] and out["edge_ok"]
    if not out["confirmed"]:
        out["reason"] = "edge not qualified"
    return out


# ───────────────────────── Notification ─────────────────────────
def _fmt_px(x):
    if x is None:
        return "—"
    if x >= 100:
        return f"{x:,.2f}"
    if x >= 1:
        return f"{x:.3f}"
    return f"{x:.6f}"


def format_signal(s):
    arrow = "🟢 LONG" if s["dir"] == "long" else "🔴 SHORT"
    lines = [f"{arrow}  *{s['symbol']}*"]
    tag = s.get("conf_tag") or "—"
    cf = s.get("conf_confluence") or "—"
    lines.append(f"• Confluence: {tag} · {cf}  | sources: {'+'.join(s['sources'])}")
    lines.append(f"• Forecast: 15m {s['move15']:+.2f}% ✅  1h {s['move1h']:+.2f}% ✅")
    for tf, e, q in (("15m", s["edge15"], s["q15"]), ("1h", s["edge1h"], s["q1h"])):
        if not e:
            lines.append(f"• Edge {tf}: —")
            continue
        fresh = " 🔔now" if e["sig_bars_ago"] == 0 else (
            f" ({e['sig_bars_ago']}c ago)" if e["sig_bars_ago"] > 0 else "")
        seg = (f"• Edge {tf}: {'✅' if q else '⚠️'} Exp {e['exp_R']:+.2f}R "
               f"CI[{e['ci_lo']:+.2f},{e['ci_hi']:+.2f}] n={e['n']}{fresh}")
        if q and e["sig_bars_ago"] == 0 and e["entry"]:
            seg += (f"\n    Entry {_fmt_px(e['entry'])} · Stop {_fmt_px(e['stop'])} "
                    f"· Tgt {_fmt_px(e['target'])}")
        lines.append(seg)
    return "\n".join(lines)


def format_summary(cands_eval, n_cands, n_short):
    """Compact per-cycle summary: top candidates + kahan tak pahunche."""
    lines = [f"📊 Candidates: {n_cands} · deep-checked: {n_short}"]
    top = sorted(cands_eval, key=lambda e: e["gates_passed"], reverse=True)[:8]
    for e in top:
        dirar = "🟢L" if e["dir"] == "long" else "🔴S"
        fc = (f"FC15 {'✅' if e['pass15'] else '✗'} 1h {'✅' if e['pass1h'] else '✗'}")
        if e["edge15"] is not None or e["edge1h"] is not None:
            ed = f" · Edge15 {'✅' if e['q15'] else '✗'} 1h {'✅' if e['q1h'] else '✗'}"
        else:
            ed = ""
        cf = e.get("conf_confluence") or "—"
        lines.append(f"• {dirar} {e['symbol']} [{cf}] {fc}{ed}  ({e['gates_passed']}/4)")
    return "\n".join(lines)


def telegram_send(cfg, text):
    tg = cfg.get("telegram", {})
    token = tg.get("bot_token") or os.getenv("TG_TOKEN", "")
    chat = tg.get("chat_id") or os.getenv("TG_CHAT_ID", "")
    if not token or not chat:
        log("Telegram creds missing — message print kar raha hoon:\n" + text)
        return False
    url = f"https://api.telegram.org/bot{token}/sendMessage"

    def _send(body):
        # NOTE: _req/urllib raises HTTPError on 4xx (Telegram returns 400 when
        # Markdown can't be parsed), so a failed Markdown send lands here as an
        # exception — must catch and report rather than rely on ok=False.
        try:
            _, j = post_json(url, body, timeout=30)
            return bool(j.get("ok"))
        except Exception as e:
            detail = str(e)
            try:
                detail = e.read().decode()  # HTTPError body has Telegram's reason
            except Exception:
                pass
            log(f"Telegram send error: {detail}")
            return False

    # 1) try pretty (Markdown); 2) fall back to plain text (robust vs special chars)
    if _send({"chat_id": chat, "text": text, "parse_mode": "Markdown",
              "disable_web_page_preview": True}):
        return True
    log("Telegram Markdown failed -> retrying as plain text.")
    return _send({"chat_id": chat, "text": text,
                  "disable_web_page_preview": True})


# ───────────────────────── Main ─────────────────────────
def main():
    cfg = load_config()
    state = load_state()
    cooldown = cfg.get("notify_cooldown_s", 3600)
    now = time.time()

    log("=== Confluence agent cycle START ===")
    conf_rows = run_confluence(cfg)
    scr_rows = run_screener(cfg)

    cands = build_candidates(cfg, conf_rows, scr_rows)
    log(f"Candidates merged: {len(cands)}")
    picks = shortlist(cfg, cands)
    log(f"Shortlisted for deep verify: {len(picks)} → "
        f"{[c['symbol'] for c in picks]}")

    evaluated = []
    for c in picks:
        e = evaluate_candidate(cfg, c)
        evaluated.append(e)
        mark = "✅ CONFIRMED" if e["confirmed"] else f"– {e['reason']}"
        log(f"  {e['symbol']} {e['dir']} {mark} ({e['gates_passed']}/4)")
    signals = [e for e in evaluated if e["confirmed"]]

    # dedup
    fresh = []
    for s in signals:
        key = f"{s['symbol']}_{s['dir']}"
        last = state["notified"].get(key, 0)
        if now - last >= cooldown:
            fresh.append(s)
            state["notified"][key] = now
    # prune old dedup entries
    state["notified"] = {k: v for k, v in state["notified"].items()
                         if now - v < cooldown * 6}
    save_state(state)

    ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
    if fresh:
        header = (f"🎯 *Confluence signals* — {cfg['tf_fast']}/{cfg['tf_slow']} "
                  f"({len(fresh)})\n_{ts}_\n")
        body = "\n\n".join(format_signal(s) for s in fresh)
        telegram_send(cfg, header + "\n" + body)
        log(f"Notified {len(fresh)} signal(s).")
    elif signals:
        log(f"{len(signals)} confirmed but all in cooldown — no notify.")
    else:
        log("No fully-confirmed signals this cycle.")

    # Per-cycle summary (alag se, taake dikhe agent zinda hai + kya kareeb hai)
    if cfg.get("notify_summary") and not fresh:
        summary = format_summary(evaluated, len(cands), len(picks))
        telegram_send(cfg, f"📊 *Cycle summary* — {cfg['tf_fast']}/{cfg['tf_slow']}\n"
                           f"_{ts}_\n{summary}")
        log("Summary sent.")

    log("=== cycle END ===")


if __name__ == "__main__":
    try:
        main()
    except Exception:
        log("FATAL:\n" + traceback.format_exc())
        sys.exit(1)