Case
8638500190
- Built deep-scraper skill (CDP-based X feed extraction) - Three-stage pipeline: scrape → triage → investigate - Paper trading simulator with position tracking - First live investigation: verified kch123 Polymarket profile ($9.3M P&L) - Opened first paper position: Seahawks Super Bowl @ 68c - Telegram alerts with inline action buttons - Portal build in progress (night shift)
345 lines
12 KiB
Python
Executable File
345 lines
12 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Paper trading simulator for Feed Hunter strategies.
|
|
|
|
Tracks virtual positions, P&L, and performance metrics.
|
|
No real money — everything is simulated.
|
|
|
|
Usage:
|
|
python3 simulator.py status # Show all active sims
|
|
python3 simulator.py open <strategy> <details_json> # Open a paper position
|
|
python3 simulator.py close <sim_id> <exit_price> # Close a position
|
|
python3 simulator.py update <sim_id> <current_price> # Update mark-to-market
|
|
python3 simulator.py history # Show closed positions
|
|
python3 simulator.py stats # Performance summary
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
DATA_DIR = Path(__file__).parent / "data" / "simulations"
|
|
ACTIVE_FILE = DATA_DIR / "active.json"
|
|
HISTORY_FILE = DATA_DIR / "history.json"
|
|
CONFIG_FILE = Path(__file__).parent / "config.json"
|
|
|
|
|
|
def load_config():
|
|
with open(CONFIG_FILE) as f:
|
|
return json.load(f)
|
|
|
|
|
|
def load_active():
|
|
if ACTIVE_FILE.exists():
|
|
with open(ACTIVE_FILE) as f:
|
|
return json.load(f)
|
|
return {"positions": [], "bankroll_used": 0}
|
|
|
|
|
|
def save_active(data):
|
|
DATA_DIR.mkdir(parents=True, exist_ok=True)
|
|
with open(ACTIVE_FILE, "w") as f:
|
|
json.dump(data, f, indent=2)
|
|
|
|
|
|
def load_history():
|
|
if HISTORY_FILE.exists():
|
|
with open(HISTORY_FILE) as f:
|
|
return json.load(f)
|
|
return {"closed": []}
|
|
|
|
|
|
def save_history(data):
|
|
DATA_DIR.mkdir(parents=True, exist_ok=True)
|
|
with open(HISTORY_FILE, "w") as f:
|
|
json.dump(data, f, indent=2)
|
|
|
|
|
|
def cmd_open(args):
|
|
"""Open a new paper position."""
|
|
config = load_config()
|
|
sim_config = config["simulation"]
|
|
active = load_active()
|
|
|
|
details = json.loads(args.details)
|
|
|
|
# Calculate position size
|
|
bankroll = sim_config["default_bankroll"]
|
|
max_pos = bankroll * sim_config["max_position_pct"]
|
|
position_size = details.get("size", max_pos)
|
|
position_size = min(position_size, max_pos)
|
|
|
|
sim_id = str(uuid.uuid4())[:8]
|
|
|
|
position = {
|
|
"id": sim_id,
|
|
"strategy": args.strategy,
|
|
"opened_at": datetime.now(timezone.utc).isoformat(),
|
|
"type": details.get("type", "long"), # long, short, bet
|
|
"asset": details.get("asset", "unknown"),
|
|
"entry_price": details.get("entry_price", 0),
|
|
"size": position_size,
|
|
"quantity": details.get("quantity", 0),
|
|
"stop_loss": details.get("stop_loss"),
|
|
"take_profit": details.get("take_profit"),
|
|
"current_price": details.get("entry_price", 0),
|
|
"unrealized_pnl": 0,
|
|
"unrealized_pnl_pct": 0,
|
|
"source_post": details.get("source_post", ""),
|
|
"thesis": details.get("thesis", ""),
|
|
"notes": details.get("notes", ""),
|
|
"updates": [],
|
|
}
|
|
|
|
active["positions"].append(position)
|
|
active["bankroll_used"] = sum(p["size"] for p in active["positions"])
|
|
save_active(active)
|
|
|
|
print(f"✅ Paper position opened: {sim_id}")
|
|
print(f" Strategy: {args.strategy}")
|
|
print(f" Asset: {position['asset']}")
|
|
print(f" Type: {position['type']}")
|
|
print(f" Entry: ${position['entry_price']}")
|
|
print(f" Size: ${position_size:.2f}")
|
|
if position["stop_loss"]:
|
|
print(f" Stop Loss: ${position['stop_loss']}")
|
|
if position["take_profit"]:
|
|
print(f" Take Profit: ${position['take_profit']}")
|
|
|
|
|
|
def cmd_close(args):
|
|
"""Close a paper position."""
|
|
active = load_active()
|
|
history = load_history()
|
|
|
|
pos = None
|
|
for i, p in enumerate(active["positions"]):
|
|
if p["id"] == args.sim_id:
|
|
pos = active["positions"].pop(i)
|
|
break
|
|
|
|
if not pos:
|
|
print(f"❌ Position {args.sim_id} not found")
|
|
sys.exit(1)
|
|
|
|
exit_price = float(args.exit_price)
|
|
entry_price = pos["entry_price"]
|
|
|
|
if pos["type"] == "long":
|
|
pnl_pct = (exit_price - entry_price) / entry_price if entry_price else 0
|
|
elif pos["type"] == "short":
|
|
pnl_pct = (entry_price - exit_price) / entry_price if entry_price else 0
|
|
elif pos["type"] == "bet":
|
|
# For binary bets: exit_price is 1 (win) or 0 (lose)
|
|
pnl_pct = (exit_price - entry_price) / entry_price if entry_price else 0
|
|
else:
|
|
pnl_pct = 0
|
|
|
|
realized_pnl = pos["size"] * pnl_pct
|
|
|
|
pos["closed_at"] = datetime.now(timezone.utc).isoformat()
|
|
pos["exit_price"] = exit_price
|
|
pos["realized_pnl"] = round(realized_pnl, 2)
|
|
pos["realized_pnl_pct"] = round(pnl_pct * 100, 2)
|
|
|
|
history["closed"].append(pos)
|
|
active["bankroll_used"] = sum(p["size"] for p in active["positions"])
|
|
|
|
save_active(active)
|
|
save_history(history)
|
|
|
|
emoji = "🟢" if realized_pnl >= 0 else "🔴"
|
|
print(f"{emoji} Position closed: {pos['id']}")
|
|
print(f" Asset: {pos['asset']}")
|
|
print(f" Entry: ${entry_price} → Exit: ${exit_price}")
|
|
print(f" P&L: ${realized_pnl:+.2f} ({pnl_pct*100:+.1f}%)")
|
|
|
|
|
|
def cmd_update(args):
|
|
"""Update mark-to-market for a position."""
|
|
active = load_active()
|
|
|
|
for pos in active["positions"]:
|
|
if pos["id"] == args.sim_id:
|
|
current = float(args.current_price)
|
|
entry = pos["entry_price"]
|
|
|
|
if pos["type"] == "long":
|
|
pnl_pct = (current - entry) / entry if entry else 0
|
|
elif pos["type"] == "short":
|
|
pnl_pct = (entry - current) / entry if entry else 0
|
|
else:
|
|
pnl_pct = (current - entry) / entry if entry else 0
|
|
|
|
pos["current_price"] = current
|
|
pos["unrealized_pnl"] = round(pos["size"] * pnl_pct, 2)
|
|
pos["unrealized_pnl_pct"] = round(pnl_pct * 100, 2)
|
|
pos["updates"].append({
|
|
"time": datetime.now(timezone.utc).isoformat(),
|
|
"price": current,
|
|
"pnl": pos["unrealized_pnl"],
|
|
})
|
|
|
|
# Check stop loss
|
|
if pos.get("stop_loss") and pos["type"] == "long" and current <= pos["stop_loss"]:
|
|
print(f"⚠️ STOP LOSS triggered for {pos['id']} at ${current}")
|
|
if pos.get("take_profit") and pos["type"] == "long" and current >= pos["take_profit"]:
|
|
print(f"🎯 TAKE PROFIT hit for {pos['id']} at ${current}")
|
|
|
|
save_active(active)
|
|
|
|
emoji = "🟢" if pos["unrealized_pnl"] >= 0 else "🔴"
|
|
print(f"{emoji} Updated {pos['id']}: ${current} ({pos['unrealized_pnl_pct']:+.1f}%)")
|
|
return
|
|
|
|
print(f"❌ Position {args.sim_id} not found")
|
|
|
|
|
|
def cmd_status(args):
|
|
"""Show all active positions."""
|
|
active = load_active()
|
|
config = load_config()
|
|
bankroll = config["simulation"]["default_bankroll"]
|
|
|
|
if not active["positions"]:
|
|
print("No active paper positions.")
|
|
return
|
|
|
|
total_unrealized = 0
|
|
print(f"=== Active Paper Positions ===")
|
|
print(f"Bankroll: ${bankroll} | Used: ${active['bankroll_used']:.2f} | Free: ${bankroll - active['bankroll_used']:.2f}\n")
|
|
|
|
for pos in active["positions"]:
|
|
emoji = "🟢" if pos["unrealized_pnl"] >= 0 else "🔴"
|
|
total_unrealized += pos["unrealized_pnl"]
|
|
print(f"{emoji} [{pos['id']}] {pos['strategy']}")
|
|
print(f" {pos['asset']} | {pos['type']} | Size: ${pos['size']:.2f}")
|
|
print(f" Entry: ${pos['entry_price']} → Current: ${pos['current_price']}")
|
|
print(f" P&L: ${pos['unrealized_pnl']:+.2f} ({pos['unrealized_pnl_pct']:+.1f}%)")
|
|
print(f" Opened: {pos['opened_at'][:16]}")
|
|
if pos.get("thesis"):
|
|
print(f" Thesis: {pos['thesis'][:80]}")
|
|
print()
|
|
|
|
print(f"Total unrealized P&L: ${total_unrealized:+.2f}")
|
|
|
|
|
|
def cmd_history(args):
|
|
"""Show closed positions."""
|
|
history = load_history()
|
|
|
|
if not history["closed"]:
|
|
print("No closed positions yet.")
|
|
return
|
|
|
|
print("=== Closed Positions ===\n")
|
|
for pos in history["closed"]:
|
|
emoji = "🟢" if pos["realized_pnl"] >= 0 else "🔴"
|
|
print(f"{emoji} [{pos['id']}] {pos['strategy']}")
|
|
print(f" {pos['asset']} | ${pos['entry_price']} → ${pos['exit_price']}")
|
|
print(f" P&L: ${pos['realized_pnl']:+.2f} ({pos['realized_pnl_pct']:+.1f}%)")
|
|
print(f" {pos['opened_at'][:16]} → {pos['closed_at'][:16]}")
|
|
print()
|
|
|
|
|
|
def cmd_stats(args):
|
|
"""Performance summary across all closed trades."""
|
|
history = load_history()
|
|
config = load_config()
|
|
|
|
closed = history.get("closed", [])
|
|
if not closed:
|
|
print("No completed trades to analyze.")
|
|
return
|
|
|
|
wins = [t for t in closed if t["realized_pnl"] > 0]
|
|
losses = [t for t in closed if t["realized_pnl"] <= 0]
|
|
total_pnl = sum(t["realized_pnl"] for t in closed)
|
|
|
|
print("=== Performance Summary ===\n")
|
|
print(f"Total trades: {len(closed)}")
|
|
print(f"Wins: {len(wins)} | Losses: {len(losses)}")
|
|
print(f"Win rate: {len(wins)/len(closed)*100:.1f}%")
|
|
print(f"Total P&L: ${total_pnl:+.2f}")
|
|
|
|
if wins:
|
|
avg_win = sum(t["realized_pnl"] for t in wins) / len(wins)
|
|
best = max(closed, key=lambda t: t["realized_pnl"])
|
|
print(f"Avg win: ${avg_win:+.2f}")
|
|
print(f"Best trade: {best['id']} ({best['strategy']}) ${best['realized_pnl']:+.2f}")
|
|
|
|
if losses:
|
|
avg_loss = sum(t["realized_pnl"] for t in losses) / len(losses)
|
|
worst = min(closed, key=lambda t: t["realized_pnl"])
|
|
print(f"Avg loss: ${avg_loss:+.2f}")
|
|
print(f"Worst trade: {worst['id']} ({worst['strategy']}) ${worst['realized_pnl']:+.2f}")
|
|
|
|
# ROI
|
|
bankroll = config["simulation"]["default_bankroll"]
|
|
roi = (total_pnl / bankroll) * 100
|
|
print(f"\nROI on ${bankroll} bankroll: {roi:+.1f}%")
|
|
|
|
# By strategy
|
|
strategies = {}
|
|
for t in closed:
|
|
s = t["strategy"]
|
|
if s not in strategies:
|
|
strategies[s] = {"trades": 0, "pnl": 0, "wins": 0}
|
|
strategies[s]["trades"] += 1
|
|
strategies[s]["pnl"] += t["realized_pnl"]
|
|
if t["realized_pnl"] > 0:
|
|
strategies[s]["wins"] += 1
|
|
|
|
if len(strategies) > 1:
|
|
print(f"\n=== By Strategy ===")
|
|
for name, data in sorted(strategies.items(), key=lambda x: x[1]["pnl"], reverse=True):
|
|
wr = data["wins"] / data["trades"] * 100
|
|
print(f" {name}: {data['trades']} trades, {wr:.0f}% WR, ${data['pnl']:+.2f}")
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Feed Hunter Paper Trading Simulator")
|
|
sub = parser.add_subparsers(dest="command")
|
|
|
|
sub.add_parser("status", help="Show active positions")
|
|
|
|
p_open = sub.add_parser("open", help="Open paper position")
|
|
p_open.add_argument("strategy", help="Strategy name")
|
|
p_open.add_argument("details", help="JSON with position details")
|
|
|
|
p_close = sub.add_parser("close", help="Close position")
|
|
p_close.add_argument("sim_id", help="Position ID")
|
|
p_close.add_argument("exit_price", help="Exit price")
|
|
|
|
p_update = sub.add_parser("update", help="Update mark-to-market")
|
|
p_update.add_argument("sim_id", help="Position ID")
|
|
p_update.add_argument("current_price", help="Current price")
|
|
|
|
sub.add_parser("history", help="Closed positions")
|
|
sub.add_parser("stats", help="Performance summary")
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.command == "status":
|
|
cmd_status(args)
|
|
elif args.command == "open":
|
|
cmd_open(args)
|
|
elif args.command == "close":
|
|
cmd_close(args)
|
|
elif args.command == "update":
|
|
cmd_update(args)
|
|
elif args.command == "history":
|
|
cmd_history(args)
|
|
elif args.command == "stats":
|
|
cmd_stats(args)
|
|
else:
|
|
parser.print_help()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|