Files
k 5fb1d52026 Pre-launch hardening: KOL module, Telegram, scanners, WS resilience
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>
2026-05-25 00:52:56 +08:00

215 lines
8.1 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
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()