doulfa's picture
Add bot orchestrator
08547ef verified
"""
Bot principal: orchestre les stratรฉgies, l'exรฉcution et le monitoring.
"""
import asyncio
import logging
import signal
import sys
import time
from datetime import datetime
from typing import Optional
from .config import BotConfig
from .data import GammaClient, CLOBDataClient, Market
from .execution import ExecutionEngine
from .strategies import ArbitrageStrategy, ValueBetStrategy, LeaderFollowerStrategy, Signal
from .monitor import BotMonitor, MetricSnapshot
logger = logging.getLogger("polybot")
class PolymarketBot:
"""
Bot de trading Polymarket ultra efficace.
Stratรฉgies:
1. Arbitrage intra-marchรฉ (sans risque, YES+NO < $1)
2. Value Bet (signaux de marchรฉ + Kelly sizing)
3. Leader-Follower sรฉmantique (marchรฉs corrรฉlรฉs)
Fonctionnalitรฉs:
- Gestion des risques multi-niveaux
- Mode dry-run / live
- Monitoring temps rรฉel avec alertes
- Filtrage intelligent des marchรฉs
"""
def __init__(self, config: Optional[BotConfig] = None):
self.config = config or BotConfig()
self._setup_logging()
# Composants
self.gamma = GammaClient()
self.clob = CLOBDataClient()
self.engine = ExecutionEngine(self.config)
self.monitor = BotMonitor()
# Stratรฉgies
self.strategies = {}
if "arbitrage" in self.config.strategies:
self.strategies["arbitrage"] = ArbitrageStrategy(self.config, self.clob)
if "value_bet" in self.config.strategies:
self.strategies["value_bet"] = ValueBetStrategy(self.config, self.clob)
if "leader_follower" in self.config.strategies:
self.strategies["leader_follower"] = LeaderFollowerStrategy(self.config, self.clob)
# State
self._markets: list[Market] = []
self._running = False
self._last_market_refresh = 0
self._cycle_count = 0
logger.info(f"Bot initialized | Mode: {'DRY RUN' if self.config.dry_run else 'LIVE'}")
logger.info(f"Strategies: {list(self.strategies.keys())}")
logger.info(f"Max exposure: ${self.config.max_total_exposure_usd:,.2f}")
def _setup_logging(self):
"""Configure le logging."""
log_level = getattr(logging, self.config.log_level, logging.INFO)
logging.basicConfig(
level=log_level,
format="%(asctime)s | %(name)-20s | %(levelname)-5s | %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
# โ”€โ”€ Market Management โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
async def refresh_markets(self):
"""Rafraรฎchit la liste des marchรฉs depuis Gamma API."""
logger.info("๐Ÿ”„ Refreshing markets...")
all_markets = await self.gamma.get_all_active_markets()
# Filtrer selon la config
filtered = []
for m in all_markets:
if m.volume < self.config.min_market_volume:
continue
if any(tag in self.config.excluded_tags for tag in m.tags):
continue
if m.closed or not m.active:
continue
if not m.yes_token or not m.no_token:
continue
filtered.append(m)
self._markets = filtered
self._last_market_refresh = time.time()
logger.info(f"โœ… {len(filtered)} markets loaded (from {len(all_markets)} total)")
if len(filtered) < 10:
self.monitor.alert(
"WARN",
"Low Market Count",
f"Only {len(filtered)} markets pass filters. Consider relaxing min_market_volume.",
)
def _should_refresh_markets(self) -> bool:
return (time.time() - self._last_market_refresh) > self.config.market_refresh_interval_seconds
# โ”€โ”€ Core Trading Loop โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
async def _trading_cycle(self):
"""Un cycle complet de trading: scan โ†’ signal โ†’ risque โ†’ exรฉcution."""
self._cycle_count += 1
if self._should_refresh_markets():
await self.refresh_markets()
if not self._markets:
logger.warning("No markets available, skipping cycle")
return
all_signals: list[Signal] = []
# Scanner toutes les stratรฉgies en parallรจle
scan_tasks = []
for name, strategy in self.strategies.items():
scan_tasks.append(strategy.scan(self._markets))
results = await asyncio.gather(*scan_tasks, return_exceptions=True)
for i, (name, _) in enumerate(self.strategies.items()):
if isinstance(results[i], Exception):
logger.error(f"Strategy {name} scan failed: {results[i]}")
self.monitor.alert("ERROR", f"{name} Scan Failed", str(results[i]), strategy=name)
elif results[i]:
all_signals.extend(results[i])
if not all_signals:
if self._cycle_count % 30 == 0:
logger.debug(f"Cycle {self._cycle_count}: no signals (scanning {len(self._markets)} markets)")
return
# Trier les signaux par profit attendu * confiance
all_signals.sort(key=lambda s: s.expected_profit * s.confidence, reverse=True)
# Exรฉcuter les meilleurs signaux
executed = 0
for signal in all_signals[:5]:
strategy = self.strategies.get(signal.strategy)
if not strategy:
continue
try:
trade = await strategy.execute(signal, self.engine)
if trade:
executed += 1
self.monitor.alert(
"INFO",
f"Trade Executed ({signal.strategy})",
f"{signal.action} | Confidence: {signal.confidence:.2f} | "
f"Expected profit: ${signal.expected_profit:.2f} | "
f"Size: ${signal.size_usd:.2f}",
strategy=signal.strategy,
metadata=signal.metadata or {},
)
except Exception as e:
logger.error(f"Execution failed for signal: {e}")
self.monitor.alert("ERROR", "Execution Failed", str(e), strategy=signal.strategy)
if executed > 0:
logger.info(f"โœ… Cycle {self._cycle_count}: {executed}/{len(all_signals)} signals executed")
async def _monitor_cycle(self):
"""Cycle de monitoring: mรฉtriques, alertes, dashboard."""
summary = self.engine.get_portfolio_summary()
snapshot = MetricSnapshot(
timestamp=time.time(),
balance_usd=summary["balance_usd"],
total_exposure=summary["total_exposure"],
total_pnl=summary["total_pnl"],
daily_pnl=summary["daily_pnl"],
num_positions=summary["num_positions"],
num_trades=summary["num_trades"],
max_drawdown=float(summary["max_drawdown"].replace("%", "")) / 100,
)
self.monitor.record_metrics(snapshot)
self.monitor.check_risk_alerts(summary, self.config)
# โ”€โ”€ Main Loop โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
async def run(self, max_cycles: int = None, duration_seconds: int = None):
"""
Lance le bot.
Args:
max_cycles: Nombre max de cycles (None = infini)
duration_seconds: Durรฉe max en secondes (None = infini)
"""
self._running = True
start_time = time.time()
banner = f"""
โ•”โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•—
โ•‘ ๐Ÿค– POLYMARKET ULTRA BOT โ€” STARTED ๐Ÿค– โ•‘
โ• โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•ฃ
โ•‘ Mode: {'DRY RUN (Paper Trading)' if self.config.dry_run else 'LIVE TRADING โš ๏ธ'}
โ•‘ Strategies: {', '.join(self.strategies.keys())}
โ•‘ Max Exposure: ${self.config.max_total_exposure_usd:,.2f}
โ•‘ Poll Interval: {self.config.poll_interval_seconds}s
โ•‘ Capital: ${self.engine.portfolio.balance_usd:,.2f}
โ•‘ Started at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
โ•šโ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•"""
logger.info(banner)
await self.refresh_markets()
cycle = 0
monitor_interval = 30
try:
while self._running:
cycle += 1
if max_cycles and cycle > max_cycles:
logger.info(f"Max cycles reached ({max_cycles})")
break
if duration_seconds and (time.time() - start_time) > duration_seconds:
logger.info(f"Duration limit reached ({duration_seconds}s)")
break
try:
await self._trading_cycle()
except Exception as e:
logger.error(f"Trading cycle error: {e}")
self.monitor.alert("ERROR", "Cycle Error", str(e))
if cycle % monitor_interval == 0:
await self._monitor_cycle()
self.monitor.print_dashboard()
await asyncio.sleep(self.config.poll_interval_seconds)
except KeyboardInterrupt:
logger.info("Bot stopped by user (Ctrl+C)")
except Exception as e:
logger.error(f"Fatal error: {e}")
self.monitor.alert("ERROR", "Fatal Error", str(e))
finally:
await self.shutdown()
async def shutdown(self):
"""Arrรชt propre du bot."""
logger.info("Shutting down bot...")
self._running = False
await self._monitor_cycle()
self.monitor.print_dashboard()
self.monitor.export_metrics("bot_metrics.json")
report = self.monitor.get_performance_report()
logger.info(f"""
โ•”โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•—
โ•‘ ๐Ÿ“ˆ FINAL PERFORMANCE REPORT ๐Ÿ“ˆ โ•‘
โ• โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•ฃ
โ•‘ Runtime: {report.get('uptime_hours', 0):.2f} hours
โ•‘ Final Balance: ${report.get('current_balance', 0):,.2f}
โ•‘ Total PnL: ${report.get('total_pnl', 0):+,.2f} ({report.get('return_pct', 0):+.2f}%)
โ•‘ Max Drawdown: {report.get('max_drawdown_pct', 0):.2f}%
โ•‘ Total Trades: {report.get('total_trades', 0)}
โ•‘ PnL/Hour: ${report.get('pnl_per_hour', 0):+,.2f}
โ•‘ Alerts: {report.get('total_alerts', 0)} (โš ๏ธ{report.get('warn_alerts', 0)} โŒ{report.get('error_alerts', 0)})
โ•šโ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•""")
await self.gamma.close()
await self.clob.close()
logger.info("Bot shutdown complete โœ…")
# โ”€โ”€ Quick Start โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
@classmethod
async def quick_start(cls, dry_run: bool = True, duration: int = 3600):
"""Dรฉmarrage rapide du bot."""
config = BotConfig(dry_run=dry_run)
bot = cls(config)
await bot.run(duration_seconds=duration)
return bot