5fb1d52026
Big-picture changes since b941223:
KOL pipeline (new) — Substack/podcast/blog RSS → AI ticker extraction →
on-chain wallet diff → talks-vs-trades divergence detection. Daily polls,
19 feeds, divergence emits Post + Telegram fan-out.
Telegram push (new) — walletless free tier + wallet-linked Pro upgrade,
in-bot preference commands (/trump /btc /funding /kol /conf /quiet),
signed-envelope API for dashboard. Disconnect-wallet keeps free
subscription.
BTC funding-rate reversal scanner (new) — hourly cron, 30d cumulative
funding threshold + mean-revert + 7d price confirm, emits via
/api/signals/ingest. BTC bottom-reversal scanner promoted to System 2.
WS broadcast rewrite — per-client send timeout + parallel fan-out
(asyncio.gather). Fixes "Binance WS no close frame" reconnect storms +
APScheduler 11-min job misses, both caused by one slow client stalling
the kline loop.
Error visibility — three silent-error sites (trumpstruth/truth_social
fetchers, funding_reversal scanner) now include exception type name so
httpx ConnectError-style empty-message errors stop logging blank lines.
Telegram bot loop now classifies ReadTimeout vs network vs unknown +
logger.exception for the unknown bucket.
Security hygiene — trumpsignal.db untracked from git (held subscriber
wallets + encrypted HL keys + 22 bot trades); .gitignore now blocks
*.db/.next/backups. CORS only allows FRONTEND_URL in production.
New ops scripts —
- scripts/preflight.py: env/DB/Telegram/AI auth verification gate
- scripts/backup_db.sh: cron-friendly daily DB backup (SQLite + Postgres)
- scripts/seed_kol_wallets.py: idempotent KOL on-chain wallet seeder
15 new Alembic migrations (007-021) covering convex strategy fields,
phase-1 safety, two-system frozen exits, invalidation prices, dynamic
SYS2 leverage, staged de-risk + pyramiding, peak gain tracking, risk
mode, auto-trade + grow flags, KOL module, KOL on-chain, KOL divergence,
Telegram bindings + walletless.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
215 lines
8.1 KiB
Python
215 lines
8.1 KiB
Python
"""
|
||
Past-7-day backtest under user-specified params:
|
||
- Margin: $100, leverage 20x → notional $2000 per trade
|
||
- SL: -30% on margin = -1.5% on notional move
|
||
- TP: "exit at the future peak" (look-ahead — see caveat at end)
|
||
- Window: 1 hour after publish
|
||
- Fees: 9 bps round-trip (HL taker × 2)
|
||
|
||
We only have the price_impact_m5/m15/m1h fields, which store the MAXIMUM
|
||
FAVORABLE EXCURSION (MFE). We do NOT have the trough / max ADVERSE excursion
|
||
(MAE), so we cannot detect a real intra-hour SL hit. That makes the
|
||
optimistic sim a CEILING, not a real-world outcome.
|
||
|
||
We compute three scenarios so you can see the honest spread:
|
||
|
||
A) "Perfect-foresight": exit at MFE peak. SL ignored (assume MAE = 0).
|
||
This is what the user asked for. Upper bound only.
|
||
|
||
B) "Pessimistic": if MFE < 0, assume SL hit (-1.5% × $2000 = -$30).
|
||
If MFE ≥ 0, exit at MFE peak.
|
||
Closer to reality but still optimistic on the wins.
|
||
|
||
C) "Realistic fixed TP": TP at +1.5%, SL at -1.5% (symmetric).
|
||
If MFE ≥ TP → win. Else → unknown end-of-window
|
||
price, conservative model: take MFE × 0.5 as exit.
|
||
This is the closest to a real bot's behavior.
|
||
"""
|
||
import sqlite3
|
||
import statistics
|
||
from pathlib import Path
|
||
|
||
DB = Path(__file__).resolve().parents[1] / "trumpsignal.db"
|
||
|
||
NOTIONAL = 2000.0
|
||
MARGIN = 100.0
|
||
SL_PCT = 1.5 # 1.5% notional move = -30% margin
|
||
FEE_BPS_RT = 0.0009 # 9 bps × $2000 = $1.80 per trade
|
||
|
||
|
||
def fetch_week():
|
||
"""Last 7 days of actionable signals, anchored to the LATEST post in the
|
||
DB (not wall-clock now) so the script works regardless of how stale the
|
||
snapshot is."""
|
||
con = sqlite3.connect(DB)
|
||
con.row_factory = sqlite3.Row
|
||
latest = con.execute("SELECT MAX(published_at) FROM posts").fetchone()[0]
|
||
rows = con.execute("""
|
||
SELECT id, signal, ai_confidence, price_at_post,
|
||
price_impact_m5, price_impact_m15, price_impact_m1h,
|
||
published_at, text
|
||
FROM posts
|
||
WHERE signal IN ('buy', 'short') -- exclude sell (semantic bug)
|
||
AND price_at_post IS NOT NULL
|
||
AND price_impact_m1h IS NOT NULL
|
||
AND published_at >= datetime(?, '-7 days')
|
||
ORDER BY published_at
|
||
""", (latest,)).fetchall()
|
||
con.close()
|
||
return [dict(r) for r in rows], latest
|
||
|
||
|
||
def trade_pnl_usd(gross_pct: float) -> float:
|
||
"""Convert a notional % move into $ PnL on $2000 notional, after fees."""
|
||
return NOTIONAL * (gross_pct / 100.0) - NOTIONAL * FEE_BPS_RT
|
||
|
||
|
||
def sim_perfect(rows):
|
||
"""A) Exit at peak. No SL. (User's request — look-ahead bias.)"""
|
||
pnls = []
|
||
for r in rows:
|
||
peak = r["price_impact_m1h"] # already side-adjusted, %
|
||
# Even peak < 0 still "exits at peak" per user spec
|
||
pnls.append(trade_pnl_usd(peak))
|
||
return pnls
|
||
|
||
|
||
def sim_pessimistic(rows):
|
||
"""B) If peak < 0, assume SL hit. Else exit at peak."""
|
||
pnls = []
|
||
for r in rows:
|
||
peak = r["price_impact_m1h"]
|
||
if peak < 0:
|
||
pnls.append(trade_pnl_usd(-SL_PCT)) # SL = -$30 + fees
|
||
else:
|
||
pnls.append(trade_pnl_usd(peak))
|
||
return pnls
|
||
|
||
|
||
def sim_fixed_tp(rows, tp_pct=1.5):
|
||
"""C) TP=tp_pct, SL=-1.5%. Real-bot behavior."""
|
||
pnls = []
|
||
for r in rows:
|
||
peak = r["price_impact_m1h"]
|
||
if peak >= tp_pct:
|
||
pnls.append(trade_pnl_usd(tp_pct)) # TP hit
|
||
elif peak < 0:
|
||
pnls.append(trade_pnl_usd(-SL_PCT)) # SL likely
|
||
else:
|
||
# Held the hour, peak was below TP → exit somewhere between
|
||
# 0 and peak. Use peak/2 as a midpoint estimate (conservative).
|
||
pnls.append(trade_pnl_usd(peak / 2.0))
|
||
return pnls
|
||
|
||
|
||
def stats(pnls, label):
|
||
n = len(pnls)
|
||
if n == 0:
|
||
return None
|
||
wins = sum(1 for p in pnls if p > 0)
|
||
losses = sum(1 for p in pnls if p < 0)
|
||
total = sum(pnls)
|
||
biggest_win = max(pnls)
|
||
biggest_loss = min(pnls)
|
||
win_rate = wins / n * 100
|
||
# Margin return: total $ / total margin used
|
||
total_margin_used = MARGIN * n
|
||
roi_on_margin = (total / total_margin_used) * 100
|
||
return {
|
||
"label": label,
|
||
"n": n,
|
||
"wins": wins,
|
||
"losses": losses,
|
||
"win_rate_pct": round(win_rate, 1),
|
||
"total_usd": round(total, 2),
|
||
"avg_per_trade_usd": round(total / n, 2),
|
||
"biggest_win_usd": round(biggest_win, 2),
|
||
"biggest_loss_usd": round(biggest_loss, 2),
|
||
"roi_on_margin_pct": round(roi_on_margin, 1),
|
||
}
|
||
|
||
|
||
def print_stats(s):
|
||
if s is None:
|
||
print(" (no trades)")
|
||
return
|
||
print(f" n trades: {s['n']}")
|
||
print(f" win rate: {s['win_rate_pct']}% ({s['wins']}W / {s['losses']}L)")
|
||
print(f" total PnL (USD): ${s['total_usd']:+,.2f}")
|
||
print(f" avg per trade: ${s['avg_per_trade_usd']:+,.2f}")
|
||
print(f" biggest win: ${s['biggest_win_usd']:+,.2f}")
|
||
print(f" biggest loss: ${s['biggest_loss_usd']:+,.2f}")
|
||
print(f" ROI on margin: {s['roi_on_margin_pct']:+.1f}% "
|
||
f"(${s['total_usd']:+.0f} on ${MARGIN * s['n']:.0f} total margin used)")
|
||
|
||
|
||
def main():
|
||
rows, anchor = fetch_week()
|
||
print("=" * 72)
|
||
print(f"PAST-7-DAY BACKTEST — TrumpSignal AI strategy")
|
||
print(f"(7 days back from latest DB post: {anchor[:19]})")
|
||
print("=" * 72)
|
||
print(f"Position size: ${MARGIN} margin × 20x leverage = ${NOTIONAL:.0f} notional")
|
||
print(f"Stop loss: -30% margin = -{SL_PCT}% notional")
|
||
print(f"Take profit: see scenarios below")
|
||
print(f"Fees: 9 bps round-trip = ${NOTIONAL * FEE_BPS_RT:.2f} per trade")
|
||
print(f"Hold window: 1 hour")
|
||
print(f"Sample: {len(rows)} actionable signals (buy/short, sell excluded)")
|
||
if rows:
|
||
print(f"Date span: {rows[0]['published_at'][:10]} → {rows[-1]['published_at'][:10]}")
|
||
print()
|
||
|
||
if not rows:
|
||
print("⚠️ No actionable signals in the last 7 days.")
|
||
print(" Either Trump didn't post anything market-relevant, or all signals")
|
||
print(" were 'hold'. Cannot run backtest.")
|
||
return
|
||
|
||
print("=" * 72)
|
||
print("SCENARIO A — 'Perfect foresight' (exit at peak, ignore SL) ← user spec")
|
||
print(" This is the THEORETICAL CEILING. No real bot can hit this.")
|
||
print("=" * 72)
|
||
print_stats(stats(sim_perfect(rows), "perfect"))
|
||
print()
|
||
|
||
print("=" * 72)
|
||
print("SCENARIO B — 'Pessimistic' (SL hits if peak<0, else exit at peak)")
|
||
print(" Closer to honest, but wins are still cherry-picked at peak.")
|
||
print("=" * 72)
|
||
print_stats(stats(sim_pessimistic(rows), "pessimistic"))
|
||
print()
|
||
|
||
print("=" * 72)
|
||
print("SCENARIO C — 'Realistic fixed TP/SL' (TP=+1.5%, SL=-1.5%)")
|
||
print(" Closest to what a real bot with these params would yield.")
|
||
print(" THIS is the only one defensible for marketing.")
|
||
print("=" * 72)
|
||
print_stats(stats(sim_fixed_tp(rows, tp_pct=1.5), "realistic"))
|
||
print()
|
||
|
||
# Also try a couple of TP variations to find sweet spot
|
||
print("=" * 72)
|
||
print("Realistic sim — TP threshold sweep (SL fixed at -1.5%)")
|
||
print("=" * 72)
|
||
print(f" {'TP%':>6} {'n':>4} {'win%':>6} {'total$':>10} {'avg$':>8} {'ROI%margin':>11}")
|
||
for tp in [0.5, 1.0, 1.5, 2.0, 3.0]:
|
||
s = stats(sim_fixed_tp(rows, tp), f"tp={tp}")
|
||
print(f" {tp:>5.1f}% {s['n']:>4} {s['win_rate_pct']:>5.1f}% "
|
||
f"${s['total_usd']:>+8.2f} ${s['avg_per_trade_usd']:>+6.2f} "
|
||
f"{s['roi_on_margin_pct']:>+9.1f}%")
|
||
|
||
print()
|
||
print("=" * 72)
|
||
print("BOTTOM LINE")
|
||
print("=" * 72)
|
||
perfect = stats(sim_perfect(rows), "")
|
||
realistic = stats(sim_fixed_tp(rows, 1.5), "")
|
||
print(f" Perfect-foresight ceiling: ${perfect['total_usd']:+,.2f} "
|
||
f"(do NOT put on homepage — look-ahead bias)")
|
||
print(f" Realistic fixed TP=1.5%: ${realistic['total_usd']:+,.2f} "
|
||
f"(this is the marketable number, IF positive)")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|