| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972 |
- from __future__ import annotations
- import time
- 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 = "Grid Trader"
- STRATEGY_PROFILE = {
- "expects": {
- "trend": "none",
- "volatility": "low",
- "event_risk": "low",
- "liquidity": "normal",
- },
- "avoids": {
- "trend": "strong",
- "volatility": "expanding",
- "event_risk": "high",
- "liquidity": "thin",
- },
- "risk_profile": "medium",
- "capabilities": ["structure_harvesting", "range_making", "liquidity_harvesting"],
- }
- TICK_MINUTES = 0.50
- CONFIG_SCHEMA = {
- "grid_levels": {"type": "int", "default": 6, "min": 1, "max": 20},
- "grid_step_pct": {"type": "float", "default": 0.012, "min": 0.001, "max": 0.1},
- "volatility_timeframe": {"type": "string", "default": "1h"},
- "volatility_multiplier": {"type": "float", "default": 0.5, "min": 0.0, "max": 10.0},
- "grid_step_min_pct": {"type": "float", "default": 0.005, "min": 0.0001, "max": 0.5},
- "grid_step_max_pct": {"type": "float", "default": 0.03, "min": 0.0001, "max": 1.0},
- "order_size": {"type": "float", "default": 0.0, "min": 0.0},
- "recenter_pct": {"type": "float", "default": 0.05, "min": 0.0, "max": 0.5},
- "recenter_atr_multiplier": {"type": "float", "default": 0.35, "min": 0.0, "max": 10.0},
- "recenter_min_pct": {"type": "float", "default": 0.0025, "min": 0.0, "max": 0.5},
- "recenter_max_pct": {"type": "float", "default": 0.03, "min": 0.0, "max": 0.5},
- "fee_rate": {"type": "float", "default": 0.0025, "min": 0.0, "max": 0.05},
- "trade_sides": {"type": "string", "default": "both"},
- "max_notional_per_order": {"type": "float", "default": 0.0, "min": 0.0},
- "dust_collect": {"type": "bool", "default": False},
- "order_call_delay_ms": {"type": "int", "default": 250, "min": 0, "max": 10000},
- "debug_orders": {"type": "bool", "default": True},
- }
- STATE_SCHEMA = {
- "center_price": {"type": "float", "default": 0.0},
- "last_price": {"type": "float", "default": 0.0},
- "seeded": {"type": "bool", "default": False},
- "last_action": {"type": "string", "default": "idle"},
- "last_error": {"type": "string", "default": ""},
- "orders": {"type": "list", "default": []},
- "order_ids": {"type": "list", "default": []},
- "debug_log": {"type": "list", "default": []},
- "base_available": {"type": "float", "default": 0.0},
- "counter_available": {"type": "float", "default": 0.0},
- "regimes_updated_at": {"type": "string", "default": ""},
- "account_snapshot_updated_at": {"type": "string", "default": ""},
- "last_balance_log_signature": {"type": "string", "default": ""},
- "last_balance_log_at": {"type": "string", "default": ""},
- "grid_refresh_pending_until": {"type": "string", "default": ""},
- "mismatch_ticks": {"type": "int", "default": 0},
- "recovery_cooldown_until": {"type": "string", "default": ""},
- }
- def init(self):
- return {
- "center_price": 0.0,
- "last_price": 0.0,
- "seeded": False,
- "last_action": "idle",
- "last_error": "",
- "orders": [],
- "order_ids": [],
- "debug_log": ["init cancel all orders"],
- "base_available": 0.0,
- "counter_available": 0.0,
- "regimes_updated_at": "",
- "account_snapshot_updated_at": "",
- "last_balance_log_signature": "",
- "last_balance_log_at": "",
- "grid_refresh_pending_until": "",
- "mismatch_ticks": 0,
- "recovery_cooldown_until": "",
- }
- 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("grid", message)
- def _log_decision(self, action: str, **fields) -> None:
- parts = [action]
- for key, value in fields.items():
- parts.append(f"{key}={value}")
- self._log(", ".join(parts))
- def _set_grid_refresh_pause(self, seconds: float = 30.0) -> None:
- self.state["grid_refresh_pending_until"] = (datetime.now(timezone.utc).timestamp() + max(seconds, 0.0))
- def _grid_refresh_paused(self) -> bool:
- try:
- until = float(self.state.get("grid_refresh_pending_until") or 0.0)
- except Exception:
- until = 0.0
- return until > datetime.now(timezone.utc).timestamp()
- def _recovery_paused(self) -> bool:
- try:
- until = float(self.state.get("recovery_cooldown_until") or 0.0)
- except Exception:
- until = 0.0
- return until > datetime.now(timezone.utc).timestamp()
- def _trip_recovery_pause(self, seconds: float = 30.0) -> None:
- self.state["recovery_cooldown_until"] = (datetime.now(timezone.utc).timestamp() + max(seconds, 0.0))
- def _recover_grid(self, price: float) -> None:
- self._log(f"recovery mode: cancel all and rebuild from {price}")
- try:
- self.context.cancel_all_orders()
- except Exception as exc:
- self.state["last_error"] = str(exc)
- self._log(f"recovery cancel-all failed: {exc}")
- self.state["orders"] = []
- self.state["order_ids"] = []
- self.state["open_order_count"] = 0
- self.state["center_price"] = price
- self.state["seeded"] = True
- self._place_grid(price)
- self._sync_open_orders_state()
- self.state["mismatch_ticks"] = 0
- self._trip_recovery_pause()
- def _order_count_mismatch(self, tracked_ids: list[str], live_orders: list[dict]) -> bool:
- live_ids = [str(order.get("bitstamp_order_id") or order.get("order_id") or order.get("id") or order.get("client_order_id") or "") for order in live_orders if isinstance(order, dict)]
- live_ids = [oid for oid in live_ids if oid]
- if len(live_ids) != len([oid for oid in tracked_ids if oid]):
- return True
- return False
- 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 _live_fee_rates(self) -> tuple[float, float]:
- try:
- payload = self.context.get_fee_rates(self._market_symbol())
- maker = float(payload.get("maker") or 0.0)
- taker = float(payload.get("taker") or 0.0)
- return maker, taker
- except Exception as exc:
- self._log(f"fee lookup failed: {exc}")
- fallback = float(self.config.get("fee_rate", 0.0025) or 0.0)
- return fallback, fallback
- def _live_fee_rate(self) -> float:
- _maker, taker = self._live_fee_rates()
- return taker
- def _mode(self) -> str:
- return getattr(self.context, "mode", "active") or "active"
- 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()
- step_map = {"cautious": 0.008, "normal": 0.012, "assertive": 0.018}
- recenter_map = {"cautious": 0.035, "normal": 0.05, "assertive": 0.07}
- levels_map = {"cautious": 4, "normal": 6, "assertive": 8}
- delay_map = {"cautious": 500, "normal": 250, "assertive": 120}
- if priority in {"low", "background"}:
- risk = "cautious"
- elif priority in {"high", "urgent"}:
- risk = "assertive"
- self.config["grid_step_pct"] = step_map.get(risk, 0.012)
- self.config["recenter_pct"] = recenter_map.get(risk, 0.05)
- self.config["grid_levels"] = levels_map.get(risk, 6)
- self.config["order_call_delay_ms"] = delay_map.get(risk, 250)
- self.state["policy_derived"] = {
- "grid_step_pct": self.config["grid_step_pct"],
- "recenter_pct": self.config["recenter_pct"],
- "grid_levels": self.config["grid_levels"],
- "order_call_delay_ms": self.config["order_call_delay_ms"],
- }
- return policy
- def _price(self) -> float:
- payload = self.context.get_price(self._base_symbol())
- return float(payload.get("price") or 0.0)
- def _regime_snapshot(self) -> dict:
- timeframes = ["1d", "4h", "1h", "15m"]
- snapshot = {}
- for tf in timeframes:
- try:
- snapshot[tf] = self.context.get_regime(self._base_symbol(), tf)
- except Exception as exc:
- snapshot[tf] = {"error": str(exc)}
- return snapshot
- def _refresh_regimes(self) -> None:
- self.state["regimes"] = self._regime_snapshot()
- self.state["regimes_updated_at"] = datetime.now(timezone.utc).isoformat()
- def _recenter_threshold_pct(self) -> float:
- base_threshold = float(self.config.get("recenter_pct", 0.05) or 0.05)
- atr_multiplier = float(self.config.get("recenter_atr_multiplier", 0.35) or 0.0)
- min_threshold = float(self.config.get("recenter_min_pct", 0.0025) or 0.0)
- max_threshold = float(self.config.get("recenter_max_pct", 0.03) or 1.0)
- try:
- tf = str(self.config.get("volatility_timeframe", "1h") or "1h")
- regime = self.context.get_regime(self._base_symbol(), tf)
- short_regime = self.context.get_regime(self._base_symbol(), "15m")
- atr_pct = float((regime or {}).get("volatility", {}).get("atr_percent") or 0.0)
- short_atr_pct = float((short_regime or {}).get("volatility", {}).get("atr_percent") or 0.0)
- atr_pct = max(atr_pct, short_atr_pct)
- except Exception:
- atr_pct = 0.0
- threshold = (atr_pct / 100.0) * atr_multiplier if atr_pct > 0 else base_threshold
- threshold = max(threshold, min_threshold)
- threshold = min(threshold, max_threshold)
- self.state["recenter_pct_live"] = threshold
- self.state["recenter_atr_percent"] = atr_pct
- return threshold
- def _grid_step_pct(self) -> float:
- base_step = float(self.config.get("grid_step_pct", 0.012) or 0.012)
- tf = str(self.config.get("volatility_timeframe", "1h") or "1h")
- multiplier = float(self.config.get("volatility_multiplier", 0.5) or 0.0)
- min_step = float(self.config.get("grid_step_min_pct", 0.005) or 0.0)
- max_step = float(self.config.get("grid_step_max_pct", 0.03) or 1.0)
- try:
- regime = self.context.get_regime(self._base_symbol(), tf)
- short_regime = self.context.get_regime(self._base_symbol(), "15m")
- tf_atr_pct = float((regime or {}).get("volatility", {}).get("atr_percent") or 0.0)
- atr_pct = float((regime or {}).get("volatility", {}).get("atr_percent") or 0.0)
- short_atr_pct = float((short_regime or {}).get("volatility", {}).get("atr_percent") or 0.0)
- atr_pct = max(atr_pct, short_atr_pct)
- self.state["regimes"] = self._regime_snapshot()
- except Exception as exc:
- self._log(f"regime fetch failed: {exc}")
- tf_atr_pct = 0.0
- atr_pct = 0.0
- short_atr_pct = 0.0
- adaptive = (atr_pct / 100.0) * multiplier if atr_pct > 0 else base_step
- step = adaptive if atr_pct > 0 else base_step
- step = max(step, min_step)
- step = min(step, max_step)
- self.state["grid_step_pct"] = step
- self.state["atr_percent_tf"] = tf_atr_pct
- self.state["atr_percent_15m"] = short_atr_pct
- self.state["atr_percent"] = atr_pct
- return step
- def _config_warning(self) -> str | None:
- recenter_pct = float(self.state.get("recenter_pct_live") or self._recenter_threshold_pct())
- grid_step_pct = float(self.state.get("grid_step_pct") or self._grid_step_pct())
- if grid_step_pct <= 0:
- return None
- ratio = recenter_pct / grid_step_pct
- # If the recenter threshold is too close to the first step, the grid
- # can keep rebuilding before it has a fair chance to trade.
- if ratio <= 1.0:
- return f"warning: recenter threshold ({recenter_pct:.4f}) is <= grid step ({grid_step_pct:.4f}), it may recenter before trading"
- if ratio < 1.5:
- return f"warning: recenter threshold ({recenter_pct:.4f}) is only {ratio:.2f}x the grid step ({grid_step_pct:.4f}), consider widening it"
- return None
- def _available_balance(self, asset_code: str) -> float:
- try:
- info = self.context.get_account_info()
- except Exception as exc:
- self._log(f"account info failed: {exc}")
- return 0.0
- balances = info.get("balances") if isinstance(info, dict) else []
- if not isinstance(balances, list):
- return 0.0
- wanted = str(asset_code or "").upper()
- for balance in balances:
- if not isinstance(balance, dict):
- continue
- if str(balance.get("asset_code") or "").upper() != wanted:
- continue
- try:
- return float(balance.get("available") if balance.get("available") is not None else balance.get("total") or 0.0)
- except Exception:
- return 0.0
- return 0.0
- def _refresh_balance_snapshot(self) -> None:
- try:
- info = self.context.get_account_info()
- except Exception as exc:
- self._log(f"balance refresh failed: {exc}")
- return
- balances = info.get("balances") if isinstance(info, dict) else []
- if not isinstance(balances, list):
- return
- base = self._base_symbol()
- quote = self.context.counter_currency or "USD"
- for balance in balances:
- if not isinstance(balance, dict):
- continue
- asset = str(balance.get("asset_code") or "").upper()
- try:
- available = float(balance.get("available") if balance.get("available") is not None else balance.get("total") or 0.0)
- except Exception:
- continue
- if asset == base:
- self.state["base_available"] = available
- if asset == str(quote).upper():
- self.state["counter_available"] = available
- self.state["account_snapshot_updated_at"] = datetime.now(timezone.utc).isoformat()
- signature = f"{base}:{self.state.get('base_available', 0.0):.8f}|{quote}:{self.state.get('counter_available', 0.0):.8f}"
- last_signature = str(self.state.get("last_balance_log_signature") or "")
- last_logged_at = str(self.state.get("last_balance_log_at") or "")
- now_iso = self.state["account_snapshot_updated_at"]
- should_log = signature != last_signature or not last_logged_at
- if not should_log:
- try:
- from datetime import datetime as _dt
- elapsed = (_dt.fromisoformat(now_iso) - _dt.fromisoformat(last_logged_at)).total_seconds()
- should_log = elapsed >= 60
- except Exception:
- should_log = True
- if should_log:
- self.state["last_balance_log_signature"] = signature
- self.state["last_balance_log_at"] = now_iso
- self._log_decision(
- "balance snapshot",
- base=base,
- base_available=f"{self.state.get('base_available', 0.0):.6g}",
- quote=quote,
- quote_available=f"{self.state.get('counter_available', 0.0):.6g}",
- updated_at=now_iso,
- )
- def _supported_levels(self, side: str, price: float, min_notional: float, *, balance_total: float | None = None) -> int:
- if min_notional <= 0 or price <= 0:
- return 0
- safety = 0.995
- fee_rate = self._live_fee_rate()
- if side == "buy":
- quote = self.context.counter_currency or "USD"
- quote_available = self._available_balance(quote)
- self.state["counter_available"] = quote_available
- usable_notional = (quote_available if balance_total is None else balance_total) * safety
- return max(int(usable_notional / min_notional), 0)
- base = self._base_symbol()
- base_available = self._available_balance(base)
- self.state["base_available"] = base_available
- usable_base = base_available if balance_total is None else balance_total
- usable_notional = usable_base * safety * price / (1 + fee_rate)
- return max(int(usable_notional / min_notional), 0)
- def _side_allowed(self, side: str) -> bool:
- selected = str(self.config.get("trade_sides", "both") or "both").strip().lower()
- if selected == "both":
- return True
- return selected == side
- def _desired_sides(self) -> set[str]:
- selected = str(self.config.get("trade_sides", "both") or "both").strip().lower()
- if selected == "both":
- return {"buy", "sell"}
- if selected in {"buy", "sell"}:
- return {selected}
- return {"buy", "sell"}
- def _suggest_amount(self, side: str, price: float, levels: int, min_notional: float) -> float:
- return self.context.suggest_order_amount(
- side=side,
- price=price,
- levels=levels,
- min_notional=min_notional,
- fee_rate=self._live_fee_rate(),
- max_notional_per_order=float(self.config.get("max_notional_per_order", 0.0) or 0.0),
- dust_collect=bool(self.config.get("dust_collect", False)),
- order_size=float(self.config.get("order_size", 0.0) or 0.0),
- )
- def _target_levels_for_side(self, side: str, center: float, live_orders: list[dict], balance_total: float, expected_levels: int, min_notional: float) -> int:
- if expected_levels <= 0 or center <= 0 or balance_total <= 0:
- return 0
- side_orders = [
- order for order in live_orders
- if isinstance(order, dict) and str(order.get("side") or "").lower() == side
- ]
- amount = 0.0
- if side_orders:
- amounts = []
- for order in side_orders:
- try:
- amounts.append(float(order.get("amount") or 0.0))
- except Exception:
- continue
- if amounts:
- amount = sum(amounts) / len(amounts)
- if amount <= 0:
- amount = self._suggest_amount(side, center, max(expected_levels, 1), min_notional)
- if amount <= 0:
- return 0
- fee_rate = self._live_fee_rate()
- safety = 0.995
- step = self._grid_step_pct()
- spendable_total = balance_total * safety
- for level_count in range(expected_levels, 0, -1):
- feasible = True
- if side == "buy":
- needed = 0.0
- for i in range(1, level_count + 1):
- level_price = center * (1 - (step * i))
- min_size = (min_notional / level_price) if level_price > 0 and min_notional > 0 else 0.0
- if amount < min_size:
- feasible = False
- break
- needed += amount * level_price * (1 + fee_rate)
- else:
- needed = amount * level_count
- for i in range(1, level_count + 1):
- level_price = center * (1 + (step * i))
- min_size = (min_notional / level_price) if level_price > 0 and min_notional > 0 else 0.0
- if amount < min_size:
- feasible = False
- break
- if feasible and needed <= spendable_total + 1e-9:
- return level_count
- return 0
- def _place_grid(self, center: float) -> None:
- center = self._maybe_refresh_center(center)
- mode = self._mode()
- levels = int(self.config.get("grid_levels", 6) or 6)
- step = self._grid_step_pct()
- min_notional = float(self.context.minimum_order_value or 0.0)
- market = self._market_symbol()
- orders = []
- order_ids = []
- def _capture_order_id(result):
- if isinstance(result, dict):
- return result.get("bitstamp_order_id") or result.get("order_id") or result.get("id") or result.get("client_order_id")
- return None
- buy_levels = min(levels, self._supported_levels("buy", center, min_notional)) if (mode == "active" and self._side_allowed("buy")) else (levels if self._side_allowed("buy") else 0)
- sell_levels = min(levels, self._supported_levels("sell", center, min_notional)) if (mode == "active" and self._side_allowed("sell")) else (levels if self._side_allowed("sell") else 0)
- buy_amount = self._suggest_amount("buy", center, max(buy_levels, 1), min_notional)
- sell_amount = self._suggest_amount("sell", center, max(sell_levels, 1), min_notional)
- for i in range(1, levels + 1):
- buy_price = round(center * (1 - (step * i)), 8)
- sell_price = round(center * (1 + (step * i)), 8)
- if mode != "active":
- orders.append({"side": "buy", "price": buy_price, "amount": buy_amount, "result": {"simulated": True}})
- orders.append({"side": "sell", "price": sell_price, "amount": sell_amount, "result": {"simulated": True}})
- self._log(f"plan level {i}: buy {buy_price} amount {buy_amount:.6g} / sell {sell_price} amount {sell_amount:.6g}")
- continue
- if i > buy_levels and i > sell_levels:
- self._log(f"skip level {i}: no capacity on either side")
- continue
- min_size_buy = (min_notional / buy_price) if buy_price > 0 else 0.0
- min_size_sell = (min_notional / sell_price) if sell_price > 0 else 0.0
- try:
- if i <= buy_levels and buy_amount >= min_size_buy:
- buy = self.context.place_order(side="buy", order_type="limit", amount=buy_amount, price=buy_price, market=market)
- orders.append({"side": "buy", "price": buy_price, "amount": buy_amount, "result": buy})
- buy_id = _capture_order_id(buy)
- if buy_id is not None:
- order_ids.append(str(buy_id))
- if i <= sell_levels and sell_amount >= min_size_sell:
- sell = self.context.place_order(side="sell", order_type="limit", amount=sell_amount, price=sell_price, market=market)
- orders.append({"side": "sell", "price": sell_price, "amount": sell_amount, "result": sell})
- sell_id = _capture_order_id(sell)
- if sell_id is not None:
- order_ids.append(str(sell_id))
- self._log(f"seed level {i}: buy {buy_price} amount {buy_amount:.6g} / sell {sell_price} amount {sell_amount:.6g}")
- except Exception as exc: # best effort for first draft
- self.state["last_error"] = str(exc)
- self._log(f"seed level {i} failed: {exc}")
- continue
- delay = max(int(self.config.get("order_call_delay_ms", 250) or 0), 0) / 1000.0
- if delay > 0:
- time.sleep(delay)
- self._refresh_balance_snapshot()
- self.state["orders"] = orders
- self.state["order_ids"] = order_ids
- self.state["last_action"] = "seeded grid"
- self._set_grid_refresh_pause()
- def _top_up_grid(self, center: float, live_orders: list[dict]) -> list[str]:
- center = self._maybe_refresh_center(center)
- levels = int(self.config.get("grid_levels", 6) or 6)
- if levels <= 0 or center <= 0:
- return []
- min_notional = float(self.context.minimum_order_value or 0.0)
- step = self._grid_step_pct()
- market = self._market_symbol()
- placed: list[str] = []
- live_by_side: dict[str, int] = {"buy": 0, "sell": 0}
- for order in live_orders:
- if not isinstance(order, dict):
- continue
- side = str(order.get("side") or "").lower()
- if side in live_by_side:
- live_by_side[side] += 1
- for side in ("buy", "sell"):
- live_count = live_by_side.get(side, 0)
- max_levels = self._supported_levels(side, center, min_notional)
- target_levels = min(levels, max_levels)
- if target_levels <= live_count:
- continue
- amount = self._suggest_amount(side, center, max(target_levels, 1), min_notional)
- for i in range(live_count + 1, target_levels + 1):
- price = round(center * (1 - (step * i)) if side == "buy" else center * (1 + (step * i)), 8)
- min_size = (min_notional / price) if price > 0 else 0.0
- if amount < min_size:
- self._log_decision(
- f"skip top-up {side} level {i}",
- reason="below_min_size",
- amount=f"{amount:.6g}",
- min_size=f"{min_size:.6g}",
- price=price,
- )
- continue
- try:
- self._log_decision(f"top-up {side} level {i}", price=price, amount=f"{amount:.6g}")
- result = self.context.place_order(side=side, order_type="limit", amount=amount, price=price, market=market)
- order_id = None
- if isinstance(result, dict):
- order_id = result.get("bitstamp_order_id") or result.get("order_id") or result.get("id") or result.get("client_order_id")
- if order_id is not None:
- placed.append(str(order_id))
- live_orders.append({"side": side, "price": price, "amount": amount, "result": result})
- self._refresh_balance_snapshot()
- except Exception as exc:
- self.state["last_error"] = str(exc)
- self._log_decision(f"top-up {side} level {i} failed", error=str(exc))
- continue
- delay = max(int(self.config.get("order_call_delay_ms", 250) or 0), 0) / 1000.0
- if delay > 0:
- time.sleep(delay)
- return placed
- def _recenter_and_rebuild_from_fill(self, fill_price: float) -> None:
- fill_price = self._maybe_refresh_center(fill_price)
- """Treat a fill as the new market anchor and rebuild the full grid from there."""
- if fill_price <= 0:
- return
- self._recenter_and_rebuild_from_price(fill_price, "fill rebuild")
- def _recenter_and_rebuild_from_price(self, price: float, reason: str) -> None:
- if price <= 0:
- return
- current = float(self.state.get("center_price") or 0.0)
- self._log(f"{reason}: recenter from {current} to {price}")
- try:
- self.context.cancel_all_orders()
- except Exception as exc:
- self.state["last_error"] = str(exc)
- self._log(f"{reason} cancel-all failed: {exc}")
- # Give the exchange a moment to release balance before we rebuild.
- time.sleep(3.0)
- self._refresh_balance_snapshot()
- self.state["center_price"] = price
- self.state["seeded"] = True
- self._place_grid(price)
- self._refresh_balance_snapshot()
- self._set_grid_refresh_pause()
- def on_stop(self):
- self._log("stopping: cancel all open orders")
- try:
- self.context.cancel_all_orders()
- except Exception as exc:
- self.state["last_error"] = str(exc)
- self._log(f"stop cancel-all failed: {exc}")
- self.state["orders"] = []
- self.state["order_ids"] = []
- self.state["open_order_count"] = 0
- self.state["last_action"] = "stopped"
- def _maybe_refresh_center(self, price: float) -> float:
- if price <= 0:
- return price
- current = float(self.state.get("center_price") or 0.0)
- if current <= 0:
- self.state["center_price"] = price
- return price
- deviation = abs(price - current) / current if current else 0.0
- threshold = self._recenter_threshold_pct()
- if deviation >= threshold:
- self._log(f"recenter anchor from {current} to {price} dev={deviation:.4f} threshold={threshold:.4f}")
- self.state["center_price"] = price
- return price
- return current
- def _sync_open_orders_state(self) -> list[dict]:
- try:
- open_orders = self.context.get_open_orders()
- except Exception as exc:
- self.state["last_error"] = str(exc)
- self._log(f"open orders sync failed: {exc}")
- return []
- if not isinstance(open_orders, list):
- open_orders = []
- live_orders = [order for order in open_orders if isinstance(order, dict)]
- live_ids = [str(order.get("bitstamp_order_id") or order.get("order_id") or order.get("id") or order.get("client_order_id") or "") for order in live_orders]
- live_ids = [oid for oid in live_ids if oid]
- live_sides = [str(order.get("side") or "").lower() for order in live_orders]
- self.state["orders"] = live_orders
- self.state["order_ids"] = live_ids
- self.state["open_order_count"] = len(live_ids)
- self._log(f"sync live orders: count={len(live_ids)} sides={live_sides} ids={live_ids}")
- return live_orders
- def _cancel_orders(self, order_ids) -> None:
- for order_id in order_ids or []:
- self._log(f"dropping stale order {order_id} from state")
- def _reconcile_after_sync(self, previous_orders: list[dict], live_orders: list[dict], desired_sides: set[str], price: float) -> tuple[list[dict], list[str], int]:
- live_ids = list(self.state.get("order_ids") or [])
- open_order_count = len(live_ids)
- if self._mode() != "active":
- return live_orders, live_ids, open_order_count
- previous_ids = {
- str(order.get("bitstamp_order_id") or order.get("order_id") or order.get("id") or order.get("client_order_id") or "")
- for order in previous_orders
- if isinstance(order, dict)
- }
- current_ids = {
- str(order.get("bitstamp_order_id") or order.get("order_id") or order.get("id") or order.get("client_order_id") or "")
- for order in live_orders
- if isinstance(order, dict)
- }
- vanished_orders = [
- order
- for order in previous_orders
- if isinstance(order, dict)
- and str(order.get("bitstamp_order_id") or order.get("order_id") or order.get("id") or order.get("client_order_id") or "") in (previous_ids - current_ids)
- ]
- if vanished_orders and not self._grid_refresh_paused():
- for order in vanished_orders:
- order_id = str(order.get("bitstamp_order_id") or order.get("order_id") or order.get("id") or order.get("client_order_id") or "")
- if not order_id:
- continue
- try:
- payload = self.context.query_order(order_id)
- except Exception as exc:
- self._log(f"order status query failed for {order_id}: {exc}")
- continue
- raw = payload.get("raw") if isinstance(payload, dict) else {}
- if not isinstance(raw, dict):
- raw = {}
- status = str(payload.get("status") or raw.get("status") or order.get("status") or "").strip().lower()
- if status in {"finished", "filled", "closed"}:
- fill_price = 0.0
- for candidate in (raw.get("price"), order.get("price"), price):
- try:
- fill_price = float(candidate or 0.0)
- except Exception:
- fill_price = 0.0
- if fill_price > 0:
- break
- if fill_price > 0:
- self._log(f"filled order {order_id} detected via exec status={status}, recentering at {fill_price}")
- self._recenter_and_rebuild_from_fill(fill_price)
- live_orders = self._sync_open_orders_state()
- live_ids = list(self.state.get("order_ids") or [])
- open_order_count = len(live_ids)
- return live_orders, live_ids, open_order_count
- if status in {"cancelled", "expired", "missing"}:
- self._log(f"vanished order {order_id} resolved as {status}")
- continue
- return live_orders, live_ids, open_order_count
- def on_tick(self, tick):
- previous_orders = list(self.state.get("orders") or [])
- tracked_ids_before_sync = list(self.state.get("order_ids") or [])
- rebuild_done = False
- self._refresh_balance_snapshot()
- price = self._price()
- self.state["last_price"] = price
- self.state["last_error"] = ""
- self._refresh_regimes()
- try:
- live_orders = self._sync_open_orders_state()
- live_ids = list(self.state.get("order_ids") or [])
- open_order_count = len(live_ids)
- expected_ids = [str(oid) for oid in tracked_ids_before_sync if oid]
- stale_ids = []
- missing_ids = []
- except Exception as exc:
- open_order_count = -1
- live_orders = []
- live_ids = []
- expected_ids = []
- stale_ids = []
- missing_ids = []
- self.state["last_error"] = str(exc)
- self._log(f"open orders check failed: {exc}")
- self.state["open_order_count"] = open_order_count
- desired_sides = self._desired_sides()
- mode = self._mode()
- if mode != "active":
- if open_order_count > 0:
- self._log("observe mode: cancel all open orders")
- try:
- self.context.cancel_all_orders()
- except Exception as exc:
- self.state["last_error"] = str(exc)
- self._log(f"observe cancel failed: {exc}")
- self.state["orders"] = []
- self.state["order_ids"] = []
- self.state["open_order_count"] = 0
- if not self.state.get("seeded") or not self.state.get("center_price"):
- self.state["center_price"] = price
- self.state["seeded"] = True
- self.state["last_action"] = "observe monitor"
- self._log(f"observe at {price} dev 0.0000")
- return {"action": "observe", "price": price, "deviation": 0.0}
- center = float(self.state.get("center_price") or price)
- recenter_pct = float(self.config.get("recenter_pct", 0.05) or 0.05)
- deviation = abs(price - center) / center if center else 0.0
- if deviation >= recenter_pct:
- self.state["center_price"] = price
- self.state["last_action"] = "observe monitor"
- self._log(f"observe at {price} dev {deviation:.4f}")
- return {"action": "observe", "price": price, "deviation": deviation}
- self.state["last_action"] = "observe monitor"
- self._log(f"observe at {price} dev {deviation:.4f}")
- return {"action": "observe", "price": price, "deviation": deviation}
- if stale_ids:
- self._log(f"stale live orders: {stale_ids}")
- self._cancel_orders(stale_ids)
- live_ids = [oid for oid in live_ids if oid not in stale_ids]
- if missing_ids:
- self._log(f"missing tracked orders: {missing_ids}")
- self.state["order_ids"] = live_ids
- missing_tracked = bool(set(expected_ids) - set(live_ids))
- center = self._maybe_refresh_center(float(self.state.get("center_price") or price))
- recenter_pct = self._recenter_threshold_pct()
- deviation = abs(price - center) / center if center else 0.0
- if mode == "active" and deviation >= recenter_pct and not self._grid_refresh_paused():
- if rebuild_done:
- return {"action": "hold", "price": price}
- self._log(f"recenter needed at price={price} center={center} dev={deviation:.4f} threshold={recenter_pct:.4f}")
- rebuild_done = True
- self._recenter_and_rebuild_from_price(price, "recenter")
- live_orders = self._sync_open_orders_state()
- live_ids = list(self.state.get("order_ids") or [])
- open_order_count = len(live_ids)
- self.state["last_action"] = "recentered"
- return {"action": "recenter", "price": price, "deviation": deviation}
- live_orders, live_ids, open_order_count = self._reconcile_after_sync(previous_orders, live_orders, desired_sides, price)
- if desired_sides != {"buy", "sell"}:
- self._log("single-side mode is disabled for this strategy, forcing full-grid rebuilds only")
- current_sides = {str(order.get("side") or "").lower() for order in live_orders if isinstance(order, dict)}
- current_buy = sum(1 for order in live_orders if isinstance(order, dict) and str(order.get("side") or "").lower() == "buy")
- current_sell = sum(1 for order in live_orders if isinstance(order, dict) and str(order.get("side") or "").lower() == "sell")
- expected_levels = int(self.config.get("grid_levels", 6) or 6)
- quote_currency = self.context.counter_currency or "USD"
- quote_available = self._available_balance(quote_currency)
- base_symbol = self._base_symbol()
- base_available = self._available_balance(base_symbol)
- reserved_quote = sum(
- float(order.get("price") or 0.0) * float(order.get("amount") or 0.0)
- for order in live_orders
- if isinstance(order, dict) and str(order.get("side") or "").lower() == "buy"
- )
- reserved_base = sum(
- float(order.get("amount") or 0.0)
- for order in live_orders
- if isinstance(order, dict) and str(order.get("side") or "").lower() == "sell"
- )
- total_quote = quote_available + reserved_quote
- total_base = base_available + reserved_base
- target_buy = self._target_levels_for_side("buy", price, live_orders, total_quote, expected_levels, float(self.context.minimum_order_value or 0.0))
- target_sell = self._target_levels_for_side("sell", price, live_orders, total_base, expected_levels, float(self.context.minimum_order_value or 0.0))
- target_total = target_buy + target_sell
- grid_not_as_expected = (
- bool(live_orders)
- and (
- current_buy != target_buy
- or current_sell != target_sell
- )
- )
- can_make_better = target_total > 0 and (current_buy != target_buy or current_sell != target_sell)
- if grid_not_as_expected and can_make_better and not self._grid_refresh_paused():
- if rebuild_done:
- return {"action": "hold", "price": price}
- self._log(
- f"grid shape mismatch, rebuilding full grid: live_buy={current_buy} live_sell={current_sell} target_buy={target_buy} target_sell={target_sell}"
- )
- rebuild_done = True
- self.state["center_price"] = price
- self._recenter_and_rebuild_from_price(price, "grid shape rebuild")
- live_orders = self._sync_open_orders_state()
- mode = self._mode()
- self.state["last_action"] = "reseeded" if mode == "active" else f"{mode} monitor"
- return {"action": "reseed" if mode == "active" else "plan", "price": price}
- if grid_not_as_expected and not can_make_better:
- self._log(
- f"grid shape left unchanged, balance cannot improve it: live_buy={current_buy} live_sell={current_sell} target_buy={target_buy} target_sell={target_sell}"
- )
- if self._order_count_mismatch(tracked_ids_before_sync, live_orders):
- if rebuild_done:
- return {"action": "hold", "price": price}
- self._log(f"grid mismatch detected, rebuilding full grid: tracked={len(tracked_ids_before_sync)} live={len(live_orders)}")
- rebuild_done = True
- self.state["center_price"] = price
- self._recenter_and_rebuild_from_price(price, "grid mismatch rebuild")
- live_orders = self._sync_open_orders_state()
- mode = self._mode()
- self.state["last_action"] = "reseeded" if mode == "active" else f"{mode} monitor"
- return {"action": "reseed" if mode == "active" else "plan", "price": price}
- if (not self.state.get("seeded") or not self.state.get("center_price")) and not self._grid_refresh_paused():
- self.state["center_price"] = price
- self._place_grid(price)
- live_orders = self._sync_open_orders_state()
- self.state["seeded"] = True
- mode = self._mode()
- self._log(f"{'seeded' if mode == 'active' else 'planned'} grid at {price}")
- return {"action": "seed" if mode == "active" else "plan", "price": price}
- if ((open_order_count == 0) or missing_tracked) and not self._grid_refresh_paused():
- if rebuild_done:
- return {"action": "hold", "price": price}
- self._log("missing tracked order(s), rebuilding full grid")
- rebuild_done = True
- self.state["center_price"] = price
- self._recenter_and_rebuild_from_price(price, "missing order rebuild")
- live_orders = self._sync_open_orders_state()
- mode = self._mode()
- self.state["last_action"] = "reseeded" if mode == "active" else f"{mode} monitor"
- return {"action": "reseed" if mode == "active" else "plan", "price": price}
- mode = self._mode()
- self.state["last_action"] = "hold" if mode == "active" else f"{mode} monitor"
- self._log(f"hold at {price} dev {deviation:.4f}")
- return {"action": "hold" if mode == "active" else "plan", "price": price, "deviation": deviation}
- 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": {
- "balances": {
- "base_available": self.state.get("base_available", 0.0),
- "counter_available": self.state.get("counter_available", 0.0),
- },
- "open_orders": self.state.get("orders") or [],
- "exposure": "grid",
- },
- "state": {
- "center_price": self.state.get("center_price", 0.0),
- "last_price": self.state.get("last_price", 0.0),
- "last_action": self.state.get("last_action", "idle"),
- "open_order_count": self.state.get("open_order_count", 0),
- "regimes_updated_at": self.state.get("regimes_updated_at", ""),
- },
- "assessment": {
- "confidence": None,
- "uncertainty": None,
- "reason": "structure-based grid management",
- "warnings": [w for w in [self._config_warning()] if w],
- "policy": dict(self.config.get("policy") or {}),
- },
- "execution": snapshot.get("execution", {}),
- }
- def render(self):
- # Refresh the market-derived display values on render so the dashboard
- # reflects the same inputs the strategy would use on the next tick.
- live_step_pct = float(self.state.get("grid_step_pct") or 0.0)
- live_atr_pct = float(self.state.get("atr_percent") or 0.0)
- try:
- self._refresh_balance_snapshot()
- live_step_pct = self._grid_step_pct()
- live_atr_pct = float(self.state.get("atr_percent") or live_atr_pct)
- except Exception as exc:
- self._log(f"render refresh failed: {exc}")
- return {
- "widgets": [
- {"type": "metric", "label": "market", "value": self._market_symbol()},
- {"type": "metric", "label": "center", "value": round(float(self.state.get("center_price") or 0.0), 6)},
- {"type": "metric", "label": "last price", "value": round(float(self.state.get("last_price") or 0.0), 6)},
- {"type": "metric", "label": "state", "value": self.state.get("last_action", "idle")},
- {"type": "metric", "label": "orders", "value": len(self.state.get("orders") or [])},
- {"type": "metric", "label": "open orders", "value": self.state.get("open_order_count", 0)},
- {"type": "metric", "label": f"ATR({self.config.get('volatility_timeframe', '1h')}) %", "value": round(live_atr_pct, 4)},
- {"type": "metric", "label": "grid step %", "value": round(live_step_pct * 100.0, 4)},
- {"type": "metric", "label": "1d", "value": ((self.state.get('regimes') or {}).get('1d') or {}).get('trend', {}).get('state', 'n/a')},
- {"type": "metric", "label": "4h", "value": ((self.state.get('regimes') or {}).get('4h') or {}).get('trend', {}).get('state', 'n/a')},
- {"type": "metric", "label": "1h", "value": ((self.state.get('regimes') or {}).get('1h') or {}).get('trend', {}).get('state', 'n/a')},
- {"type": "metric", "label": "15m", "value": ((self.state.get('regimes') or {}).get('15m') or {}).get('trend', {}).get('state', 'n/a')},
- {"type": "metric", "label": f"{self._base_symbol()} avail", "value": round(float(self.state.get("base_available") or 0.0), 8)},
- {"type": "metric", "label": f"{self.context.counter_currency or 'USD'} avail", "value": round(float(self.state.get("counter_available") or 0.0), 8)},
- *([
- {"type": "text", "label": "config warning", "value": warning},
- ] if (warning := self._config_warning()) else []),
- {"type": "text", "label": "error", "value": self.state.get("last_error", "") or "none"},
- {"type": "log", "label": "debug log", "lines": self.state.get("debug_log") or []},
- ]
- }
|