Skip to content

feat(fase-19): celulas bidireccionales (IN_SHORT + exits invertidos) #7

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
guigerdts merged 1 commit into from
Jun 26, 2026
Merged

feat(fase-19): celulas bidireccionales (IN_SHORT + exits invertidos) #7

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
201 changes: 152 additions & 49 deletions cells/base.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,11 +15,16 @@
class Cell:
"""Autonomous trading cell.

Maintains internal bar history and state machine (IDLE / IN_POSITION).
Maintains internal bar history and state machine
(IDLE / IN_POSITION / IN_SHORT).
On each tick event, evaluates entry or exit rules and returns a
trading signal dict when triggered.
"""

# ── State constants ────────────────────────────────────────────────
IN_LONG = "IN_POSITION" # semantic alias for new code
IN_SHORT = "IN_SHORT" # short position state

def __init__(
self,
config: dict[str, Any],
Expand All @@ -29,7 +34,8 @@ def __init__(

Args:
config: Dict with keys ``name``, ``symbol``, ``entry``,
``exit`` (list), ``risk`` (dict with ``sizing``).
``short_entry``, ``exit`` (list),
``risk`` (dict with ``sizing``).
inference_engine: InferenceEngine instance for rule evaluation.
"""
self.name: str = config.get("name", "unnamed")
Expand All @@ -41,10 +47,11 @@ def __init__(
self.sizing: float = float(risk_cfg.get("sizing", 0.01))
self.max_positions: int = int(risk_cfg.get("max_positions", 5))

# ── Inference engine + pre-built graph ─────────────────────────
# ── Inference engine + pre-built graphs ────────────────────────
self.config: dict[str, Any] = config
self.inference_engine: Any = inference_engine
self.entry_config: dict[str, Any] = config.get("entry", {})
self.short_entry_config: dict[str, Any] = config.get("short_entry", {})

# Pre‑build the entry condition graph ONCE (Bug 6)
self._entry_graph: Any = None
Expand All @@ -55,6 +62,15 @@ def __init__(
except Exception:
logger.exception("Error construyendo grafo de entrada para {}", self.name)

# Pre‑build the short entry condition graph ONCE
self._short_entry_graph: Any = None
if self.short_entry_config and self.inference_engine:
from inference.graph import build_graph
try:
self._short_entry_graph = build_graph(self.short_entry_config)
except Exception:
logger.exception("Error construyendo grafo de entrada short para {}", self.name)

# ── Parse exit list (YAML: list of {type, params}) ─────────────
self.exit_list: list[dict[str, Any]] = config.get("exit", [])
self._parse_exit_rules()
Expand All @@ -64,6 +80,8 @@ def __init__(
self.max_bars: int = 500
self.entry_price: float = 0.0
self.position_qty: float = 0.0
self._trailing_high: float = 0.0
self._trailing_low: float = 0.0

# ── State management ─────────────────────────────────────────────

Expand All @@ -73,8 +91,8 @@ def reset_state(self) -> None:
self.bars = []
self.entry_price = 0.0
self.position_qty = 0.0
if hasattr(self, "_trailing_high"):
self._trailing_high = 0.0
self._trailing_high = 0.0
self._trailing_low = 0.0

# ── Exit rule parsing ─────────────────────────────────────────────

Expand Down Expand Up @@ -123,30 +141,37 @@ async def handle(self, event: dict[str, Any]) -> dict[str, Any] | None:
current_price = event.get("price", 0.0)

if self.state == "IDLE":
return await self._check_entry(current_price)
signal = await self._check_entry(current_price)
if signal is not None:
return signal
return await self._check_short_entry(current_price)

if self.state == "IN_POSITION":
if self.state in ("IN_POSITION", "IN_SHORT"):
return self._check_exit(current_price)

return None

# ── Entry logic ───────────────────────────────────────────────────

def enter_position(self, price: float) -> None:
"""Mark this cell as IN_POSITION after risk approval.
def enter_position(self, price: float, direction: str = "long") -> None:
"""Mark this cell as IN_POSITION or IN_SHORT after risk approval.

Called by the EventEngine AFTER the RiskManager approves the
signal, so that cells rejected by risk stay in IDLE state and
can retry.

Args:
price: Entry price.
direction: ``"long"`` (default) → ``state = "IN_POSITION"``.
``"short"`` → ``state = "IN_SHORT"``.
"""
self.state = "IN_POSITION"
self.state = "IN_SHORT" if direction == "short" else "IN_POSITION"
self.entry_price = price
self._trailing_high = 0.0
self._trailing_low = 0.0
logger.info(
"{} {} ENTRADA @ ${:.2f}",
self.symbol, self.name, price,
"{} {} ENTRADA {} @ ${:.2f}",
self.symbol, self.name, direction, price,
)

async def _check_entry(self, current_price: float) -> dict[str, Any] | None:
Expand Down Expand Up @@ -191,17 +216,60 @@ async def _check_entry(self, current_price: float) -> dict[str, Any] | None:
"cell_name": self.name, # so RiskManager can track per-cell
}

async def _check_short_entry(self, current_price: float) -> dict[str, Any] | None:
"""Evaluate short entry conditions.

Mirrors ``_check_entry()`` but reads ``self._short_entry_graph``
and returns an ``action="SHORT"`` signal.

Args:
current_price: Current market price.

Returns:
A SHORT signal dict if short entry conditions are met, or None.
"""
if not self.short_entry_config or not self.inference_engine:
return None
if self._short_entry_graph is None:
return None

market_data = self._build_data()
if not market_data:
return None

try:
should_enter = self._short_entry_graph.evaluate(market_data)
except Exception:
logger.exception("Error evaluando condiciones de entrada short para {}", self.name)
return None

if not should_enter:
return None

return {
"action": "SHORT",
"symbol": self.symbol,
"price": current_price,
"sizing": self.sizing,
"cell_name": self.name,
}

# ── Exit logic ────────────────────────────────────────────────────

def _check_exit(self, current_price: float) -> dict[str, Any] | None:
"""Evaluate exit conditions (ATR-based stop-loss, take-profit,
trailing stop, or z-score).

Thresholds are inverted when ``self.state == "IN_SHORT"`` so that
take-profit triggers when the price falls and stop-loss when it
rises (the mirror of long logic).

Args:
current_price: Current market price.

Returns:
A SELL signal if any exit condition is met, or None.
A signal dict (action="BUY" for short exits, "SELL" for
long exits) if any condition is met, or None.
"""
if self.entry_price == 0.0:
return None
Expand All @@ -211,42 +279,73 @@ def _check_exit(self, current_price: float) -> dict[str, Any] | None:
return None

atr_pct = atr / self.entry_price if self.entry_price > 0 else 0.0
is_short = self.state == "IN_SHORT"

# Stop-loss (ATR-based)
if self.exit_stop_loss is not None:
stop_price = self.entry_price * (1.0 - self.exit_stop_loss * atr_pct)
if current_price <= stop_price:
logger.info(
"{} {} STOP-LOSS @ ${:.2f} (ATR {:.2f}, entry ${:.2f})",
self.symbol, self.name, current_price, atr, self.entry_price,
)
return self._exit_signal(current_price)
if is_short:
stop_price = self.entry_price * (1.0 + self.exit_stop_loss * atr_pct)
if current_price >= stop_price:
logger.info(
"{} {} SHORT STOP-LOSS @ ${:.2f} (ATR {:.2f}, entry ${:.2f})",
self.symbol, self.name, current_price, atr, self.entry_price,
)
return self._exit_signal(current_price)
else:
stop_price = self.entry_price * (1.0 - self.exit_stop_loss * atr_pct)
if current_price <= stop_price:
logger.info(
"{} {} STOP-LOSS @ ${:.2f} (ATR {:.2f}, entry ${:.2f})",
self.symbol, self.name, current_price, atr, self.entry_price,
)
return self._exit_signal(current_price)

# Take-profit (ATR-based)
if self.exit_take_profit is not None:
take_price = self.entry_price * (1.0 + self.exit_take_profit * atr_pct)
if current_price >= take_price:
logger.info(
"{} {} TAKE-PROFIT @ ${:.2f} (ATR {:.2f}, entry ${:.2f})",
self.symbol, self.name, current_price, atr, self.entry_price,
)
return self._exit_signal(current_price)

# Trailing stop (ATR-based)
if is_short:
take_price = self.entry_price * (1.0 - self.exit_take_profit * atr_pct)
if current_price <= take_price:
logger.info(
"{} {} SHORT TAKE-PROFIT @ ${:.2f} (ATR {:.2f}, entry ${:.2f})",
self.symbol, self.name, current_price, atr, self.entry_price,
)
return self._exit_signal(current_price)
else:
take_price = self.entry_price * (1.0 + self.exit_take_profit * atr_pct)
if current_price >= take_price:
logger.info(
"{} {} TAKE-PROFIT @ ${:.2f} (ATR {:.2f}, entry ${:.2f})",
self.symbol, self.name, current_price, atr, self.entry_price,
)
return self._exit_signal(current_price)

# Trailing stop (ATR-based) — direction-aware
if self.exit_trailing_stop is not None:
trail_distance = self.exit_trailing_stop * atr
if not hasattr(self, '_trailing_high'):
self._trailing_high = current_price
self._trailing_high = max(self._trailing_high, current_price)
if current_price <= self._trailing_high - trail_distance:
logger.info(
"{} {} TRAILING-STOP @ ${:.2f} (high ${:.2f})",
self.symbol, self.name, current_price, self._trailing_high,
)
self._trailing_high = 0.0
return self._exit_signal(current_price)

# Z-score exit (reversion to mean)
if is_short:
if self._trailing_low == 0.0:
self._trailing_low = current_price
self._trailing_low = min(self._trailing_low, current_price)
if current_price >= self._trailing_low + trail_distance:
logger.info(
"{} {} SHORT TRAILING-STOP @ ${:.2f} (low ${:.2f})",
self.symbol, self.name, current_price, self._trailing_low,
)
self._trailing_low = 0.0
return self._exit_signal(current_price)
else:
if self._trailing_high == 0.0:
self._trailing_high = current_price
self._trailing_high = max(self._trailing_high, current_price)
if current_price <= self._trailing_high - trail_distance:
logger.info(
"{} {} TRAILING-STOP @ ${:.2f} (high ${:.2f})",
self.symbol, self.name, current_price, self._trailing_high,
)
self._trailing_high = 0.0
return self._exit_signal(current_price)

# Z-score exit (reversion to mean) — direction-neutral
if self.exit_zscore_threshold is not None:
market_data = self._build_data()
if market_data:
Expand All @@ -264,21 +363,25 @@ def _check_exit(self, current_price: float) -> dict[str, Any] | None:
return None

def _exit_signal(self, current_price: float) -> dict[str, Any]:
"""Generate a SELL signal WITHOUT resetting cell state.
"""Generate an exit signal WITHOUT resetting cell state.

Returns ``action="BUY"`` when the cell is in ``IN_SHORT``
(buy-to-cover) and ``action="SELL"`` otherwise (long exit).

State is reset by the engine via ``exit_position()`` AFTER
the RiskManager approves AND the broker executes the trade,
mirroring how ``enter_position()`` works for BUY signals.
mirroring how ``enter_position()`` works for entry signals.

Args:
current_price: Current market price.

Returns:
SELL signal dict with action, symbol, price, sizing,
Signal dict with action, symbol, price, sizing,
entry_price, and cell_name.
"""
action = "BUY" if self.state == "IN_SHORT" else "SELL"
return {
"action": "SELL",
"action": action,
"symbol": self.symbol,
"price": current_price,
"sizing": self.sizing,
Expand All @@ -287,16 +390,16 @@ def _exit_signal(self, current_price: float) -> dict[str, Any]:
}

def exit_position(self) -> None:
"""Mark this cell as IDLE after a successful SELL execution.
"""Mark this cell as IDLE after a successful exit.

Called by the EventEngine AFTER the RiskManager approves and
the broker executes the SELL trade. This is the mirror of
the broker executes the exit trade. This is the mirror of
``enter_position()`` — state changes only after confirmation.
"""
self.state = "IDLE"
self.entry_price = 0.0
if hasattr(self, '_trailing_high'):
self._trailing_high = 0.0
self._trailing_high = 0.0
self._trailing_low = 0.0

# ── Data helpers ──────────────────────────────────────────────────

Expand Down
Loading