george
2be8b491d0
- Grid strategy with survival-gated spacing and depth - Full 60% drop simulation for all survival checks - Calibration report with auto-updating survival threshold - Transaction history sync from Capital.com - Dip mode with bottom-two TP rules
120 lines
4.2 KiB
Python
120 lines
4.2 KiB
Python
"""
|
|
actions.py — Action handler for dry/confirm/live modes.
|
|
========================================================
|
|
Every action the bot takes (place order, cancel order, update TP)
|
|
goes through this handler. The mode controls what actually happens.
|
|
|
|
MODES:
|
|
------
|
|
dryrun → Logs what it would do. Returns None. No API calls made.
|
|
confirm → Prints the action, waits for y/n input. Only proceeds on 'y'.
|
|
live → Executes immediately, no prompts.
|
|
|
|
All three modes log the action clearly so you can audit what happened.
|
|
|
|
WHY THIS MATTERS:
|
|
-----------------
|
|
Before trusting the bot with live orders, run in dryrun and confirm modes.
|
|
Watch the logs for several days. Only switch to live when you're confident
|
|
the logic is correct and the orders it wants to place match your expectations.
|
|
"""
|
|
|
|
import logging
|
|
from typing import Optional
|
|
from client import CapitalClient
|
|
import config
|
|
|
|
log = logging.getLogger("maxbot.actions")
|
|
|
|
|
|
class ActionHandler:
|
|
|
|
def __init__(self, mode: str, client: CapitalClient):
|
|
self.mode = mode
|
|
self.client = client
|
|
self.actions_taken = 0
|
|
self.orders_placed = 0
|
|
self.orders_cancelled = 0
|
|
self.tp_updates = 0
|
|
|
|
def _log_action(self, symbol: str, description: str):
|
|
"""Print a highly visible action log line."""
|
|
log.warning(f"")
|
|
log.warning(f"{'━' * 50}")
|
|
log.warning(f" {symbol} {description}")
|
|
log.warning(f"{'━' * 50}")
|
|
|
|
def _confirm(self, description: str) -> bool:
|
|
if self.mode == "dryrun":
|
|
log.info(f"[DRY RUN] Would: {description}")
|
|
return False
|
|
if self.mode == "confirm":
|
|
print(f"\n ACTION: {description}")
|
|
try:
|
|
answer = input(" Proceed? (y/n): ").strip().lower()
|
|
except (EOFError, KeyboardInterrupt):
|
|
return False
|
|
return answer == "y"
|
|
return True # live
|
|
|
|
# ─────────────────────────────────────────
|
|
# ACTIONS
|
|
# ─────────────────────────────────────────
|
|
|
|
def place_limit_order(self, size: float, limit_price: float,
|
|
take_profit: float) -> Optional[dict]:
|
|
desc = (
|
|
f"PLACE ORDER entry ${limit_price:.2f} "
|
|
f"size {size} TP ${take_profit:.2f}"
|
|
)
|
|
if self._confirm(desc):
|
|
result = self.client.place_limit_order(size, limit_price, take_profit)
|
|
self._log_action("✚", desc)
|
|
self.actions_taken += 1
|
|
self.orders_placed += 1
|
|
return result
|
|
return None
|
|
|
|
def cancel_order(self, deal_id: str, price: float) -> Optional[dict]:
|
|
desc = f"CANCEL ORDER ${price:.2f}"
|
|
if self._confirm(desc):
|
|
result = self.client.cancel_order(deal_id)
|
|
self._log_action("✖", desc)
|
|
self.actions_taken += 1
|
|
self.orders_cancelled += 1
|
|
return result
|
|
return None
|
|
|
|
def enable_tp(self, deal_id: str, entry_price: float,
|
|
take_profit: float) -> Optional[dict]:
|
|
desc = (
|
|
f"ENABLE TP position ${entry_price:.2f} "
|
|
f"→ TP ${take_profit:.2f}"
|
|
)
|
|
if self._confirm(desc):
|
|
result = self.client.update_position_tp(deal_id, take_profit)
|
|
self._log_action("✔", desc)
|
|
self.actions_taken += 1
|
|
self.tp_updates += 1
|
|
return result
|
|
return None
|
|
|
|
def disable_tp(self, deal_id: str, entry_price: float) -> Optional[dict]:
|
|
desc = f"DISABLE TP position ${entry_price:.2f} (manual close)"
|
|
if self._confirm(desc):
|
|
result = self.client.update_position_tp(deal_id, None)
|
|
self._log_action("⊘", desc)
|
|
self.actions_taken += 1
|
|
self.tp_updates += 1
|
|
return result
|
|
return None
|
|
|
|
def summary(self) -> str:
|
|
return (
|
|
f"Session summary: "
|
|
f"{self.actions_taken} total actions | "
|
|
f"{self.orders_placed} placed | "
|
|
f"{self.orders_cancelled} cancelled | "
|
|
f"{self.tp_updates} TP updates"
|
|
)
|