""" Backtest: replay AI signals against actual price moves stored in the DB. Zero AI calls. Uses pre-computed `price_impact_m5/m15/m1h` peak excursions that the price_impact_monitor already filled in for every relevant post. Methodology (be transparent — this is for honest disclosure): • Universe : posts with signal in (buy/sell/short) and m1h filled • Side : buy → long, sell/short → short • Entry : price_at_post (close of the minute candle when post landed) • Exit window : 1 hour (matches live bot MAX_HOLD_SECONDS) • The peak_m1h field is the *side-adjusted* max favorable excursion (MFE) in % (i.e. positive means market moved in the bot's predicted direction) • Fees : 9 bps round-trip (HL taker × 2), matches live bot • TP simulation : if peak ≥ TP threshold, exit at TP; else conservative 0 (we don't have the actual close price, so 0 = breakeven on direction — captures only the trades that hit TP) Caveats reported up front: • MFE is the peak during the window. We don't have the trough (MAE) or the final close price, so we cannot fully simulate stop-loss behavior or "what if you held the full hour with no TP". This is a TP-only sim. • No slippage modeled beyond the 9 bps fee. Real fills on illiquid moments add 1-3 bps more. • Sample is whatever's in the DB right now (~70 actionable signals). """ import sqlite3 import statistics from pathlib import Path DB = Path(__file__).resolve().parents[1] / "trumpsignal.db" FEE_BPS_ROUND_TRIP = 0.0009 # 9 bps = HL taker × 2 TP_LEVELS = [0.10, 0.20, 0.30, 0.50, 1.00] # in percent def fetch_signals(): con = sqlite3.connect(DB) con.row_factory = sqlite3.Row rows = con.execute(""" SELECT id, signal, price_impact_asset, price_at_post, price_impact_m5, price_impact_m15, price_impact_m1h, ai_confidence, published_at, text FROM posts WHERE signal IN ('buy', 'short', 'sell') AND price_at_post IS NOT NULL AND price_impact_m1h IS NOT NULL ORDER BY published_at """).fetchall() con.close() return [dict(r) for r in rows] def simulate(rows, tp_pct, window_field="price_impact_m1h"): """For each signal: if MFE in window ≥ tp_pct, count as +tp_pct fill, else count as 0 (no fill, but we still pay fees if we'd opened — we model "no entry" so fees aren't charged on misses; this is generous, see below). Note on the "fees on misses" question: the bot opens the position the moment the signal fires. Even if TP isn't hit and you exit at the 1h mark, you paid the round-trip fees. So a stricter sim charges fees on EVERY trade. We do that — this is the conservative interpretation. """ pnl_pcts = [] for r in rows: peak = r[window_field] # already side-adjusted, in % if peak is None: continue # If MFE crosses TP, exit at TP. Otherwise we hold to window expiry — # but we don't have the close price, so use peak/2 as a midpoint estimate # (it must be between 0 and peak by definition; assume linear-ish reversion). if peak >= tp_pct: gross = tp_pct else: gross = peak / 2.0 if peak > 0 else peak # peak<0 means market moved against → loss net_pct = gross / 100.0 - FEE_BPS_ROUND_TRIP pnl_pcts.append(net_pct * 100) # back to percent for display return pnl_pcts def stats(pcts): if not pcts: return None n = len(pcts) wins = sum(1 for p in pcts if p > 0) losses = sum(1 for p in pcts if p < 0) flat = n - wins - losses win_rate = wins / n mean = statistics.mean(pcts) median = statistics.median(pcts) pmax = max(pcts) pmin = min(pcts) # Profit factor: sum of wins / abs(sum of losses) gross_win = sum(p for p in pcts if p > 0) gross_loss = abs(sum(p for p in pcts if p < 0)) pf = gross_win / gross_loss if gross_loss > 0 else float("inf") # Cumulative PnL (linear sum, not compounded — conservative) total = sum(pcts) return { "n": n, "wins": wins, "losses": losses, "flat": flat, "win_rate_pct": round(win_rate * 100, 1), "mean_pct": round(mean, 3), "median_pct": round(median, 3), "max_pct": round(pmax, 3), "min_pct": round(pmin, 3), "profit_factor": round(pf, 2) if pf != float("inf") else "∞", "total_pct": round(total, 2), } def main(): rows = fetch_signals() print(f"BACKTEST — TrumpSignal AI strategy on Truth Social posts") print(f"=" * 70) print(f"Universe: {len(rows)} actionable signals (buy/sell/short)") by_signal = {} for r in rows: by_signal[r['signal']] = by_signal.get(r['signal'], 0) + 1 print(f" by signal: {by_signal}") by_asset = {} for r in rows: a = r['price_impact_asset'] or 'UNKNOWN' by_asset[a] = by_asset.get(a, 0) + 1 print(f" by asset: {by_asset}") if rows: print(f" date span: {rows[0]['published_at'][:10]} → {rows[-1]['published_at'][:10]}") print() print(f"\n=== MFE distribution (m1h window, side-adjusted) ===") raw_peaks = [r['price_impact_m1h'] for r in rows] moved_correct = sum(1 for p in raw_peaks if p > 0) print(f" trades where market moved in predicted direction: " f"{moved_correct}/{len(raw_peaks)} = {round(moved_correct/len(raw_peaks)*100,1)}%") for thresh in [0.1, 0.2, 0.3, 0.5, 1.0, 2.0]: hit = sum(1 for p in raw_peaks if p >= thresh) print(f" MFE ≥ {thresh:>4.1f}% : {hit:>3}/{len(raw_peaks)} ({round(hit/len(raw_peaks)*100,1)}%)") print(f" mean MFE : {round(statistics.mean(raw_peaks),3)}%") print(f" median MFE : {round(statistics.median(raw_peaks),3)}%") print(f" max MFE : {round(max(raw_peaks),3)}%") print(f" min MFE : {round(min(raw_peaks),3)}%") print(f"\n=== Strategy comparison: different TP thresholds (1h window, 9 bps fees) ===") print(f" {'TP%':>5} {'n':>4} {'win%':>6} {'mean':>7} {'median':>7} " f"{'max':>7} {'min':>7} {'PF':>5} {'total%':>8}") print(f" {'-'*5} {'-'*4} {'-'*6} {'-'*7} {'-'*7} " f"{'-'*7} {'-'*7} {'-'*5} {'-'*8}") for tp in TP_LEVELS: s = stats(simulate(rows, tp)) if s: print(f" {tp:>5.2f} {s['n']:>4} {s['win_rate_pct']:>5.1f}% " f"{s['mean_pct']:>+6.3f}% {s['median_pct']:>+6.3f}% " f"{s['max_pct']:>+6.3f}% {s['min_pct']:>+6.3f}% " f"{str(s['profit_factor']):>5} {s['total_pct']:>+7.2f}%") # Also break down by signal direction print(f"\n=== By signal direction (TP=0.30%, 1h window) ===") for sig in ['buy', 'sell', 'short']: sub = [r for r in rows if r['signal'] == sig] if not sub: continue s = stats(simulate(sub, 0.30)) print(f" {sig:>5}: n={s['n']:>3} win_rate={s['win_rate_pct']:>5.1f}% " f"mean={s['mean_pct']:>+6.3f}% total={s['total_pct']:>+7.2f}% PF={s['profit_factor']}") # Confidence-bucket performance print(f"\n=== By AI confidence bucket (TP=0.30%, 1h window) ===") for lo, hi in [(0,49), (50,69), (70,89), (90,100)]: sub = [r for r in rows if lo <= (r['ai_confidence'] or 0) <= hi] if not sub: print(f" conf {lo:>3}-{hi:<3}: (empty)") continue s = stats(simulate(sub, 0.30)) print(f" conf {lo:>3}-{hi:<3}: n={s['n']:>3} win_rate={s['win_rate_pct']:>5.1f}% " f"mean={s['mean_pct']:>+6.3f}% total={s['total_pct']:>+7.2f}% PF={s['profit_factor']}") # Window comparison: which TP horizon best captures the move? print(f"\n=== Best window comparison (TP=0.30%) ===") print(f" Same TP, different exit windows. Tells you which horizon to trade.") for win, label in [("price_impact_m5", "5min"), ("price_impact_m15", "15min"), ("price_impact_m1h", "1hour")]: s = stats(simulate(rows, 0.30, window_field=win)) if s: print(f" {label:>6}: n={s['n']:>3} win_rate={s['win_rate_pct']:>5.1f}% " f"mean={s['mean_pct']:>+6.3f}% total={s['total_pct']:>+7.2f}% PF={s['profit_factor']}") print(f"\n{'=' * 70}") print(f"DISCLAIMER: This is a TP-only simulation using max favorable excursion") print(f" data. Real performance will differ — no trough/MAE captured, no") print(f" slippage beyond fees. Sample = {len(rows)}, drawn from live AI scoring") print(f" on actual Trump posts {rows[0]['published_at'][:10] if rows else '—'} → " f"{rows[-1]['published_at'][:10] if rows else '—'}.") if __name__ == "__main__": main()