282 lines
10 KiB
Python
282 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Strategy Performance Sentinel — Auto-run strategies, rank winners, spawn variants.
|
|
|
|
Tracks trading/prediction strategy performance over time, ranks by ROI,
|
|
and suggests parameter variations for top performers.
|
|
|
|
Usage:
|
|
python3 strategy-sentinel.py status # Show all strategy performance
|
|
python3 strategy-sentinel.py evaluate # Run evaluation cycle
|
|
python3 strategy-sentinel.py add <name> <json> # Register a strategy
|
|
python3 strategy-sentinel.py rank # Rank strategies by performance
|
|
python3 strategy-sentinel.py spawn # Auto-spawn variants of winners
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
import random
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
DATA_DIR = Path(__file__).parent.parent / "data" / "strategy-sentinel"
|
|
DATA_DIR.mkdir(parents=True, exist_ok=True)
|
|
STRATEGIES_FILE = DATA_DIR / "strategies.json"
|
|
HISTORY_FILE = DATA_DIR / "history.jsonl"
|
|
|
|
# Seed strategies based on Polymarket patterns
|
|
DEFAULT_STRATEGIES = {
|
|
"tail-conservative": {
|
|
"type": "polymarket",
|
|
"description": "Buy events under 5% probability, min $50k volume",
|
|
"params": {"threshold": 0.05, "min_volume": 50000, "position_pct": 2},
|
|
"metrics": {"trades": 0, "wins": 0, "losses": 0, "total_pnl": 0, "roi_pct": 0},
|
|
"active": True,
|
|
"created": datetime.now().isoformat(),
|
|
},
|
|
"tail-aggressive": {
|
|
"type": "polymarket",
|
|
"description": "Buy events under 15% probability, min $10k volume",
|
|
"params": {"threshold": 0.15, "min_volume": 10000, "position_pct": 5},
|
|
"metrics": {"trades": 0, "wins": 0, "losses": 0, "total_pnl": 0, "roi_pct": 0},
|
|
"active": True,
|
|
"created": datetime.now().isoformat(),
|
|
},
|
|
"momentum-high-vol": {
|
|
"type": "polymarket",
|
|
"description": "Follow momentum on high-volume markets near 50%",
|
|
"params": {"price_range": [0.40, 0.60], "min_volume": 100000, "position_pct": 3},
|
|
"metrics": {"trades": 0, "wins": 0, "losses": 0, "total_pnl": 0, "roi_pct": 0},
|
|
"active": True,
|
|
"created": datetime.now().isoformat(),
|
|
},
|
|
"crypto-sentiment-bull": {
|
|
"type": "reddit-intel",
|
|
"description": "Go long crypto when Reddit sentiment is bullish across 3+ subs",
|
|
"params": {"min_bullish_subs": 3, "hold_hours": 24},
|
|
"metrics": {"trades": 0, "wins": 0, "losses": 0, "total_pnl": 0, "roi_pct": 0},
|
|
"active": True,
|
|
"created": datetime.now().isoformat(),
|
|
},
|
|
}
|
|
|
|
|
|
def load_strategies():
|
|
if STRATEGIES_FILE.exists():
|
|
return json.loads(STRATEGIES_FILE.read_text())
|
|
save_strategies(DEFAULT_STRATEGIES)
|
|
return DEFAULT_STRATEGIES
|
|
|
|
|
|
def save_strategies(strategies):
|
|
STRATEGIES_FILE.write_text(json.dumps(strategies, indent=2))
|
|
|
|
|
|
def log_event(strategy_name, event_type, data):
|
|
entry = {
|
|
"ts": datetime.now().isoformat(),
|
|
"strategy": strategy_name,
|
|
"event": event_type,
|
|
"data": data,
|
|
}
|
|
with open(HISTORY_FILE, "a") as f:
|
|
f.write(json.dumps(entry) + "\n")
|
|
|
|
|
|
def record_trade(strategy_name, pnl, details=""):
|
|
"""Record a trade result for a strategy."""
|
|
strategies = load_strategies()
|
|
if strategy_name not in strategies:
|
|
print(f"❌ Strategy '{strategy_name}' not found")
|
|
return
|
|
|
|
s = strategies[strategy_name]
|
|
m = s["metrics"]
|
|
m["trades"] += 1
|
|
m["total_pnl"] += pnl
|
|
if pnl > 0:
|
|
m["wins"] += 1
|
|
else:
|
|
m["losses"] += 1
|
|
m["roi_pct"] = (m["total_pnl"] / max(m["trades"], 1)) * 100 # Simplified
|
|
m["win_rate"] = m["wins"] / max(m["trades"], 1)
|
|
m["last_trade"] = datetime.now().isoformat()
|
|
|
|
save_strategies(strategies)
|
|
log_event(strategy_name, "trade", {"pnl": pnl, "details": details})
|
|
print(f" 📝 Recorded: {strategy_name} → P&L: ${pnl:+.2f}")
|
|
|
|
|
|
def rank_strategies():
|
|
"""Rank all strategies by performance."""
|
|
strategies = load_strategies()
|
|
|
|
ranked = sorted(
|
|
[(name, s) for name, s in strategies.items()],
|
|
key=lambda x: x[1]["metrics"].get("roi_pct", 0),
|
|
reverse=True
|
|
)
|
|
|
|
print(f"\n🏆 Strategy Rankings — {datetime.now().strftime('%Y-%m-%d %H:%M')}")
|
|
print("=" * 70)
|
|
print(f"{'Rank':<5} {'Strategy':<25} {'Trades':<8} {'Win%':<8} {'P&L':<12} {'Status'}")
|
|
print("-" * 70)
|
|
|
|
for i, (name, s) in enumerate(ranked, 1):
|
|
m = s["metrics"]
|
|
win_rate = m.get("win_rate", 0)
|
|
status = "🟢" if s["active"] else "🔴"
|
|
medal = ["🥇", "🥈", "🥉"][i-1] if i <= 3 and m["trades"] > 0 else f" {i}."
|
|
|
|
print(f"{medal:<5} {name:<25} {m['trades']:<8} {win_rate:>5.0%} ${m['total_pnl']:>+9.2f} {status}")
|
|
|
|
if not any(s["metrics"]["trades"] > 0 for _, s in ranked):
|
|
print("\n 📭 No trades recorded yet. Strategies are seeded and ready.")
|
|
print(" Run polymarket-autopilot.py to generate trade signals.")
|
|
|
|
return ranked
|
|
|
|
|
|
def spawn_variants():
|
|
"""Auto-spawn parameter variants of top-performing strategies."""
|
|
strategies = load_strategies()
|
|
|
|
# Find strategies with trades
|
|
active_with_trades = {n: s for n, s in strategies.items()
|
|
if s["active"] and s["metrics"]["trades"] > 0}
|
|
|
|
if not active_with_trades:
|
|
# Spawn variants of seed strategies instead
|
|
print("📭 No trade data yet. Spawning variants of seed strategies...")
|
|
active_with_trades = {n: s for n, s in strategies.items() if s["active"]}
|
|
|
|
spawned = 0
|
|
for name, s in list(active_with_trades.items())[:3]:
|
|
variant_name = f"{name}-v{random.randint(100, 999)}"
|
|
if variant_name in strategies:
|
|
continue
|
|
|
|
# Mutate parameters
|
|
new_params = dict(s["params"])
|
|
for key, val in new_params.items():
|
|
if isinstance(val, (int, float)):
|
|
# Vary by ±20%
|
|
delta = val * 0.2 * random.choice([-1, 1])
|
|
new_params[key] = round(val + delta, 4)
|
|
|
|
strategies[variant_name] = {
|
|
"type": s["type"],
|
|
"description": f"Auto-variant of {name}",
|
|
"params": new_params,
|
|
"metrics": {"trades": 0, "wins": 0, "losses": 0, "total_pnl": 0, "roi_pct": 0},
|
|
"active": True,
|
|
"parent": name,
|
|
"created": datetime.now().isoformat(),
|
|
}
|
|
spawned += 1
|
|
print(f" 🧬 Spawned: {variant_name}")
|
|
print(f" Params: {json.dumps(new_params)}")
|
|
|
|
save_strategies(strategies)
|
|
print(f"\n✅ Spawned {spawned} variant(s)")
|
|
|
|
|
|
def evaluate():
|
|
"""Run evaluation cycle: check Polymarket signals against strategy params."""
|
|
strategies = load_strategies()
|
|
polymarket_data = DATA_DIR.parent / "polymarket" / "latest-signals.json"
|
|
reddit_data = DATA_DIR.parent / "reddit-intel" / "latest.json"
|
|
|
|
print(f"\n🔬 Strategy Evaluation — {datetime.now().strftime('%Y-%m-%d %H:%M')}")
|
|
print("=" * 60)
|
|
|
|
# Check Polymarket signals
|
|
if polymarket_data.exists():
|
|
signals = json.loads(polymarket_data.read_text())
|
|
print(f" 📊 Polymarket: {len(signals.get('signals', []))} signals available")
|
|
|
|
for name, s in strategies.items():
|
|
if s["type"] == "polymarket" and s["active"]:
|
|
matching = []
|
|
for sig in signals.get("signals", []):
|
|
params = s["params"]
|
|
if "threshold" in params and sig.get("price", 1) < params["threshold"]:
|
|
matching.append(sig)
|
|
if matching:
|
|
print(f" 🎯 {name}: {len(matching)} matching signal(s)")
|
|
log_event(name, "signals_matched", {"count": len(matching)})
|
|
else:
|
|
print(" ⚠️ No Polymarket data. Run polymarket-autopilot.py scan first.")
|
|
|
|
# Check Reddit sentiment
|
|
if reddit_data.exists():
|
|
intel = json.loads(reddit_data.read_text())
|
|
sentiment = intel.get("overall_market_sentiment", "neutral")
|
|
print(f" 📰 Reddit sentiment: {sentiment}")
|
|
|
|
for name, s in strategies.items():
|
|
if s["type"] == "reddit-intel" and s["active"]:
|
|
bullish_subs = sum(1 for v in intel.get("sentiment_by_sub", {}).values() if v == "bullish")
|
|
threshold = s["params"].get("min_bullish_subs", 3)
|
|
if bullish_subs >= threshold:
|
|
print(f" 🎯 {name}: SIGNAL (bullish subs: {bullish_subs}/{threshold})")
|
|
log_event(name, "signal_triggered", {"bullish_subs": bullish_subs})
|
|
else:
|
|
print(" ⚠️ No Reddit data. Run reddit-market-intel.py first.")
|
|
|
|
print("\n✅ Evaluation complete")
|
|
|
|
|
|
def show_status():
|
|
strategies = load_strategies()
|
|
print(f"\n📡 Strategy Sentinel — {len(strategies)} strategies tracked")
|
|
print("=" * 60)
|
|
for name, s in strategies.items():
|
|
status = "🟢 Active" if s["active"] else "🔴 Paused"
|
|
print(f"\n {name} [{status}]")
|
|
print(f" Type: {s['type']} | {s['description']}")
|
|
print(f" Params: {json.dumps(s['params'])}")
|
|
m = s["metrics"]
|
|
if m["trades"] > 0:
|
|
print(f" Performance: {m['trades']} trades | {m.get('win_rate', 0):.0%} win | ${m['total_pnl']:+.2f} P&L")
|
|
if s.get("parent"):
|
|
print(f" Parent: {s['parent']}")
|
|
|
|
|
|
def add_strategy(name, config_json):
|
|
strategies = load_strategies()
|
|
try:
|
|
config = json.loads(config_json)
|
|
except json.JSONDecodeError:
|
|
print(f"❌ Invalid JSON: {config_json}")
|
|
return
|
|
|
|
strategies[name] = {
|
|
"type": config.get("type", "custom"),
|
|
"description": config.get("description", "Custom strategy"),
|
|
"params": config.get("params", {}),
|
|
"metrics": {"trades": 0, "wins": 0, "losses": 0, "total_pnl": 0, "roi_pct": 0},
|
|
"active": True,
|
|
"created": datetime.now().isoformat(),
|
|
}
|
|
save_strategies(strategies)
|
|
print(f"✅ Added strategy: {name}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
cmd = sys.argv[1] if len(sys.argv) > 1 else "status"
|
|
if cmd == "status":
|
|
show_status()
|
|
elif cmd == "evaluate":
|
|
evaluate()
|
|
elif cmd == "rank":
|
|
rank_strategies()
|
|
elif cmd == "spawn":
|
|
spawn_variants()
|
|
elif cmd == "add" and len(sys.argv) >= 4:
|
|
add_strategy(sys.argv[2], sys.argv[3])
|
|
elif cmd == "record" and len(sys.argv) >= 4:
|
|
record_trade(sys.argv[2], float(sys.argv[3]), sys.argv[4] if len(sys.argv) > 4 else "")
|
|
else:
|
|
print(__doc__)
|