analysis/parse_alerts.py

#!/usr/bin/env python3
"""parse_alerts.py — extract Trade Copilot alerts from a Telegram Desktop JSON export.

Usage:  python3 parse_alerts.py result.json [--out alerts.csv]
Input:  Telegram Desktop > bot chat > Export chat history > JSON  (result.json)
Output: alerts.csv  (ts_utc, iso, symbol, direction, conf, entry, sl, tp1, tp2, tp3)

Stdlib only. Handles TG's mixed text arrays (strings + entity dicts), skips
heartbeats/tests/skip-notices, dedups same symbol+direction within 4h (keeps first).
"""
import json, re, csv, sys, datetime

DEDUP_WINDOW_S = 4 * 3600

RE_HEAD  = re.compile(r'(?:🟢|🔴)?\s*([A-Z0-9]{1,15}-USDT)\s+(LONG|SHORT)\s*·\s*([\d.]+)\s*/\s*10')
RE_ENTRY = re.compile(r'Entry\s+([\d.,]+)')
RE_SL    = re.compile(r'\bSL\s+([\d.,]+)')
RE_TP1   = re.compile(r'TP1\s+([\d.,]+)')
RE_TP2   = re.compile(r'TP2\s+([\d.,]+)')
RE_TP3   = re.compile(r'TP3\s+([\d.,]+)')

SKIP_MARKERS = ("Trade Copilot alive", "Test Message", "Test:", "Skipped", "funds in use",
                "NOT executed", "PROFIT", "LOSS", "ENTERED", "LIQUIDATED", "summary")

def flatten_text(t):
    """TG export 'text' is str OR list of str/{'text':...} — flatten to one string."""
    if isinstance(t, str):
        return t
    out = []
    for part in t:
        out.append(part if isinstance(part, str) else str(part.get("text", "")))
    return "".join(out)

def num(s):
    return float(s.replace(",", ""))

def main():
    if len(sys.argv) < 2:
        sys.exit("usage: python3 parse_alerts.py result.json [--out alerts.csv]")
    src = sys.argv[1]
    out = sys.argv[sys.argv.index("--out") + 1] if "--out" in sys.argv else "alerts.csv"

    with open(src, encoding="utf-8") as f:
        data = json.load(f)
    messages = data.get("messages", data if isinstance(data, list) else [])

    rows, skipped, last_seen = [], 0, {}
    for m in messages:
        if m.get("type") != "message":
            continue
        text = flatten_text(m.get("text", ""))
        if not text or any(k in text for k in SKIP_MARKERS):
            continue
        h = RE_HEAD.search(text)
        e, s_, t1 = RE_ENTRY.search(text), RE_SL.search(text), RE_TP1.search(text)
        if not (h and e and s_ and t1):
            continue
        # timestamp: prefer unixtime if present
        if m.get("date_unixtime"):
            ts = int(m["date_unixtime"])
        else:
            # TG export date is local time of the exporting machine; treat as-is, note in README
            ts = int(datetime.datetime.fromisoformat(m["date"]).timestamp())
        sym, direction, conf = h.group(1), h.group(2), float(h.group(3))
        key = f"{sym}:{direction}"
        if key in last_seen and ts - last_seen[key] < DEDUP_WINDOW_S:
            skipped += 1
            continue
        last_seen[key] = ts
        t2, t3 = RE_TP2.search(text), RE_TP3.search(text)
        rows.append({
            "ts_utc": ts,
            "iso": datetime.datetime.fromtimestamp(ts, datetime.timezone.utc).isoformat(),
            "symbol": sym, "direction": direction, "conf": conf,
            "entry": num(e.group(1)), "sl": num(s_.group(1)), "tp1": num(t1.group(1)),
            "tp2": num(t2.group(1)) if t2 else "", "tp3": num(t3.group(1)) if t3 else "",
        })

    rows.sort(key=lambda r: r["ts_utc"])
    with open(out, "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=list(rows[0].keys()) if rows else
                           ["ts_utc","iso","symbol","direction","conf","entry","sl","tp1","tp2","tp3"])
        w.writeheader()
        w.writerows(rows)
    print(f"parsed {len(rows)} alerts  (deduped {skipped})  →  {out}")
    if rows:
        print(f"range: {rows[0]['iso']}  →  {rows[-1]['iso']}")
        hi = [r for r in rows if r["conf"] >= 8.8]
        print(f"conf ≥ 8.8: {len(hi)} alerts")

if __name__ == "__main__":
    main()