feat: AI agent, signal engine, surge detector, portfolio simulator

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

engine/portfolio.py

@@ -0,0 +1,114 @@
+import logging
+from datetime import datetime, timezone
+from config import MAX_POSITIONS, MIN_POSITION_USD, STOP_LOSS_PCT, TAKE_PROFIT_1_PCT, TAKE_PROFIT_2_PCT
+
+logger = logging.getLogger(__name__)
+
+class PortfolioManager:
+    def __init__(self, initial_capital: float = 200.0):
+        self.initial_capital = initial_capital
+        self.cash = initial_capital
+        self.positions: dict[str, dict] = {}
+        self.trades: list[dict] = []
+
+    def buy(self, symbol: str, price: float, score: float) -> bool:
+        if symbol in self.positions:
+            return False
+        if len(self.positions) >= MAX_POSITIONS:
+            return False
+        amount = self._position_size(score)
+        if amount < MIN_POSITION_USD:
+            return False
+        if amount > self.cash:
+            amount = self.cash
+        if amount < MIN_POSITION_USD:
+            return False
+        quantity = amount / price
+        self.cash -= amount
+        self.positions[symbol] = {
+            "entry_price": price, "quantity": quantity,
+            "invested_usd": amount, "tp1_hit": False,
+            "opened_at": datetime.now(timezone.utc).isoformat(),
+        }
+        self.trades.append({
+            "coin": symbol, "side": "BUY", "price": price,
+            "quantity": quantity, "amount_usd": amount,
+            "timestamp": datetime.now(timezone.utc).isoformat(), "reason": "signal",
+        })
+        return True
+
+    def sell(self, symbol: str, price: float, reason: str = "signal", partial: float = 1.0):
+        if symbol not in self.positions:
+            return
+        pos = self.positions[symbol]
+        sell_qty = pos["quantity"] * partial
+        sell_usd = sell_qty * price
+        self.cash += sell_usd
+        self.trades.append({
+            "coin": symbol, "side": "SELL", "price": price,
+            "quantity": sell_qty, "amount_usd": sell_usd,
+            "timestamp": datetime.now(timezone.utc).isoformat(), "reason": reason,
+        })
+        if partial >= 1.0:
+            del self.positions[symbol]
+        else:
+            pos["quantity"] -= sell_qty
+
+    def check_exit(self, symbol: str, current_price: float):
+        if symbol not in self.positions:
+            return
+        pos = self.positions[symbol]
+        entry = pos["entry_price"]
+        change_pct = (current_price - entry) / entry
+        if change_pct <= STOP_LOSS_PCT:
+            self.sell(symbol, current_price, reason="stop-loss")
+            return
+        if change_pct >= TAKE_PROFIT_2_PCT:
+            self.sell(symbol, current_price, reason="take-profit-2")
+            return
+        if change_pct >= TAKE_PROFIT_1_PCT and not pos["tp1_hit"]:
+            pos["tp1_hit"] = True
+            self.sell(symbol, current_price, reason="take-profit-1", partial=0.5)
+
+    def _position_size(self, score: float) -> float:
+        if score >= 90:
+            pct = 0.30
+        elif score >= 80:
+            pct = 0.20
+        else:
+            pct = 0.15
+        return round(self.cash * pct, 2)
+
+    def get_portfolio_value(self, current_prices: dict[str, float]) -> dict:
+        holdings_value = sum(
+            pos["quantity"] * current_prices.get(sym, pos["entry_price"])
+            for sym, pos in self.positions.items()
+        )
+        total_value = self.cash + holdings_value
+        total_pnl = total_value - self.initial_capital
+        pnl_pct = (total_pnl / self.initial_capital) * 100
+
+        winning = sum(1 for t in self.trades if t["side"] == "SELL" and self._trade_pnl(t) > 0)
+        total_sells = sum(1 for t in self.trades if t["side"] == "SELL")
+        win_rate = (winning / total_sells * 100) if total_sells > 0 else 0
+
+        return {
+            "total_value": round(total_value, 2),
+            "cash": round(self.cash, 2),
+            "holdings_value": round(holdings_value, 2),
+            "total_pnl": round(total_pnl, 2),
+            "pnl_pct": round(pnl_pct, 2),
+            "win_rate": round(win_rate, 1),
+            "open_positions": len(self.positions),
+        }
+
+    def _trade_pnl(self, sell_trade: dict) -> float:
+        matching_buys = [
+            t for t in self.trades
+            if t["coin"] == sell_trade["coin"] and t["side"] == "BUY"
+            and t["timestamp"] <= sell_trade["timestamp"]
+        ]
+        if matching_buys:
+            latest_buy = matching_buys[-1]
+            return (sell_trade["price"] - latest_buy["price"]) * sell_trade["quantity"]
+        return 0