trader-mcp/strategies/trend_follower.py

from __future__ import annotations

from datetime import datetime, timezone

from src.trader_mcp.strategy_sdk import Strategy
from src.trader_mcp.logging_utils import log_event


class Strategy(Strategy):
    LABEL = "Trend Follower"
    STRATEGY_PROFILE = {
        "expects": {
            "trend": "strong",
            "volatility": "moderate",
            "event_risk": "low",
            "liquidity": "normal",
        },
        "avoids": {
            "trend": "range",
            "volatility": "chaotic",
            "event_risk": "high",
            "liquidity": "thin",
        },
        "risk_profile": "growth",
        "capabilities": ["trend_capture", "momentum_following", "position_persistence"],
    }
    TICK_MINUTES = 0.5
    CONFIG_SCHEMA = {
        "trend_timeframe": {"type": "string", "default": "1h"},
        "trend_strength_min": {"type": "float", "default": 0.65, "min": 0.0, "max": 1.0},
        "entry_offset_pct": {"type": "float", "default": 0.003, "min": 0.0, "max": 1.0},
        "exit_offset_pct": {"type": "float", "default": 0.002, "min": 0.0, "max": 1.0},
        "order_size": {"type": "float", "default": 0.0, "min": 0.0},
        "max_order_size": {"type": "float", "default": 0.0, "min": 0.0},
        "cooldown_ticks": {"type": "int", "default": 2, "min": 0, "max": 1000},
        "debug_orders": {"type": "bool", "default": True},
    }
    STATE_SCHEMA = {
        "last_price": {"type": "float", "default": 0.0},
        "last_action": {"type": "string", "default": "idle"},
        "last_error": {"type": "string", "default": ""},
        "debug_log": {"type": "list", "default": []},
        "last_signal": {"type": "string", "default": "neutral"},
        "last_strength": {"type": "float", "default": 0.0},
        "cooldown_remaining": {"type": "int", "default": 0},
        "last_order_at": {"type": "float", "default": 0.0},
        "last_order_price": {"type": "float", "default": 0.0},
    }

    def init(self):
        return {
            "last_price": 0.0,
            "last_action": "idle",
            "last_error": "",
            "debug_log": ["init trend follower"],
            "last_signal": "neutral",
            "last_strength": 0.0,
            "cooldown_remaining": 0,
            "last_order_at": 0.0,
            "last_order_price": 0.0,
        }

    def _log(self, message: str) -> None:
        state = getattr(self, "state", {}) or {}
        log = list(state.get("debug_log") or [])
        log.append(message)
        state["debug_log"] = log[-12:]
        self.state = state
        log_event("trend", message)

    def _base_symbol(self) -> str:
        return (self.context.base_currency or self.context.market_symbol or "XRP").split("/")[0].upper()

    def _market_symbol(self) -> str:
        return self.context.market_symbol or f"{self._base_symbol().lower()}usd"

    def _price(self) -> float:
        payload = self.context.get_price(self._base_symbol())
        return float(payload.get("price") or 0.0)

    def _trend_snapshot(self) -> dict:
        tf = str(self.config.get("trend_timeframe", "1h") or "1h")
        try:
            return self.context.get_regime(self._base_symbol(), tf)
        except Exception as exc:
            self._log(f"trend lookup failed: {exc}")
            return {"error": str(exc)}

    def apply_policy(self):
        policy = super().apply_policy()
        risk = str(policy.get("risk_posture") or "normal").lower()
        priority = str(policy.get("priority") or "normal").lower()

        strength_map = {"cautious": 0.8, "normal": 0.65, "assertive": 0.5}
        entry_map = {"cautious": 0.002, "normal": 0.003, "assertive": 0.005}
        exit_map = {"cautious": 0.0015, "normal": 0.002, "assertive": 0.003}
        cooldown_map = {"cautious": 4, "normal": 2, "assertive": 1}
        size_map = {"cautious": 0.5, "normal": 1.0, "assertive": 1.5}

        if priority in {"low", "background"}:
            risk = "cautious"
        elif priority in {"high", "urgent"}:
            risk = "assertive"

        self.config["trend_strength_min"] = strength_map.get(risk, 0.65)
        self.config["entry_offset_pct"] = entry_map.get(risk, 0.003)
        self.config["exit_offset_pct"] = exit_map.get(risk, 0.002)
        self.config["cooldown_ticks"] = cooldown_map.get(risk, 2)
        self.config["order_size"] = size_map.get(risk, 1.0)
        self.state["policy_derived"] = {
            "trend_strength_min": self.config["trend_strength_min"],
            "entry_offset_pct": self.config["entry_offset_pct"],
            "exit_offset_pct": self.config["exit_offset_pct"],
            "cooldown_ticks": self.config["cooldown_ticks"],
            "order_size": self.config["order_size"],
        }
        return policy

    def _trend_strength(self) -> tuple[str, float]:
        regime = self._trend_snapshot()
        trend = regime.get("trend") or {}
        direction = str(trend.get("state") or trend.get("direction") or "unknown")
        try:
            strength = float(trend.get("strength") or 0.0)
        except Exception:
            strength = 0.0
        return direction, strength

    def _suggest_amount(self, price: float) -> float:
        amount = float(self.config.get("order_size", 0.0) or 0.0)
        max_order = float(self.config.get("max_order_size", 0.0) or 0.0)
        if max_order > 0:
            amount = min(amount, max_order)
        return max(amount, 0.0)

    def on_tick(self, tick):
        self.state["last_error"] = ""
        self._log(f"tick alive price={self.state.get('last_price') or 0.0}")
        price = self._price()
        self.state["last_price"] = price

        if int(self.state.get("cooldown_remaining") or 0) > 0:
            self.state["cooldown_remaining"] = int(self.state.get("cooldown_remaining") or 0) - 1
            self.state["last_action"] = "cooldown"
            return {"action": "cooldown", "price": price}

        direction, strength = self._trend_strength()
        self.state["last_signal"] = direction
        self.state["last_strength"] = strength

        if strength < float(self.config.get("trend_strength_min", 0.65) or 0.65):
            self.state["last_action"] = "hold"
            return {"action": "hold", "price": price, "reason": "trend too weak", "strength": strength}

        amount = self._suggest_amount(price)
        if amount <= 0:
            self.state["last_action"] = "hold"
            return {"action": "hold", "price": price, "reason": "no usable size"}

        side = "buy" if direction in {"bull", "up", "long"} else "sell"
        offset = float(self.config.get("entry_offset_pct", 0.003) or 0.0)
        if side == "buy":
            order_price = round(price * (1 + offset), 8)
        else:
            order_price = round(price * (1 - offset), 8)

        try:
            if self.config.get("debug_orders", True):
                self._log(f"{side} trend amount={amount:.6g} price={order_price} strength={strength:.3f}")
            result = self.context.place_order(
                side=side,
                order_type="limit",
                amount=amount,
                price=order_price,
                market=self._market_symbol(),
            )
            self.state["cooldown_remaining"] = int(self.config.get("cooldown_ticks", 2) or 2)
            self.state["last_order_at"] = datetime.now(timezone.utc).timestamp()
            self.state["last_order_price"] = order_price
            self.state["last_action"] = f"{side}_trend"
            return {"action": side, "price": order_price, "amount": amount, "result": result, "strength": strength}
        except Exception as exc:
            self.state["last_error"] = str(exc)
            self._log(f"trend order failed: {exc}")
            self.state["last_action"] = "error"
            return {"action": "error", "price": price, "error": str(exc)}

    def report(self):
        snapshot = self.context.get_strategy_snapshot() if hasattr(self.context, "get_strategy_snapshot") else {}
        return {
            "identity": snapshot.get("identity", {}),
            "control": snapshot.get("control", {}),
            "fit": dict(getattr(self, "STRATEGY_PROFILE", {}) or {}),
            "position": snapshot.get("position", {}),
            "state": {
                "last_price": self.state.get("last_price", 0.0),
                "last_action": self.state.get("last_action", "idle"),
                "last_signal": self.state.get("last_signal", "neutral"),
                "last_strength": self.state.get("last_strength", 0.0),
                "cooldown_remaining": self.state.get("cooldown_remaining", 0),
            },
            "assessment": {
                "confidence": None,
                "uncertainty": None,
                "reason": "trend capture",
                "warnings": [],
                "policy": dict(self.config.get("policy") or {}),
            },
            "execution": snapshot.get("execution", {}),
        }

    def render(self):
        return {
            "widgets": [
                {"type": "metric", "label": "market", "value": self._market_symbol()},
                {"type": "metric", "label": "price", "value": round(float(self.state.get("last_price") or 0.0), 6)},
                {"type": "metric", "label": "signal", "value": self.state.get("last_signal", "neutral")},
                {"type": "metric", "label": "strength", "value": round(float(self.state.get("last_strength") or 0.0), 4)},
                {"type": "metric", "label": "state", "value": self.state.get("last_action", "idle")},
                {"type": "metric", "label": "cooldown", "value": int(self.state.get("cooldown_remaining") or 0)},
                {"type": "text", "label": "error", "value": self.state.get("last_error", "") or "none"},
                {"type": "log", "label": "debug log", "lines": self.state.get("debug_log") or []},
            ]
        }