biomaster/XAU_system
Watch
1
0
Fork
You've already forked XAU_system
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
XAU_system/bot.py

3570 lines
128 KiB
Python
Raw Permalink Blame History

#!/usr/bin/env python3
"""
XAUUSD five-strategy OCO trailing portfolio for MetaTrader 5.
Required user-maintained files:
1. xauusd_five_strategy_oco_bot_risk_vector_v2.py
2. timetable.csv (UTC macro-news schedule, next to this script)
3. traded_codes.csv (event-code allowlist, next to this script)
Runtime folders/files are created automatically:
logs/YYYY.txt balance changes only (UTC; initial snapshot once)
logs/YYYY_trades.log yearly closed-trade ledger (UTC)
logs/runtime.log rotating operational/error log
state/strategy_state.json restart-safe internal state
Five independent subsystems
---------------------------
1. NEWS
Scheduled events are read from timetable.csv in UTC and filtered by traded_codes.csv.
offset 2.00; initial SL 30.00; activation 5.00;
adaptive trail clip(0.25 * RMS(close changes, 5), 0.50, 2.25);
release-minute expiry 60 s; maximum hold 120 min;
risk-based volume: 6.5% of balance at the initial stop.
2. NY0830
Every weekday at 08:30 America/New_York.
offset 6.00; initial SL 60.00; activation 5.00;
adaptive trail clip(2.00 * RMS(close changes, 20), 3.00, 12.00);
at +8.00 lock +5.00; pending expiry 15 min; maximum hold 300 min;
risk-based volume: 10.5% of balance at the initial stop.
3. TOKYO0845
Every weekday at 08:45 Asia/Tokyo.
offset 3.00; initial SL 30.00; activation 3.00;
adaptive trail clip(30 * Wilder ATR(30), 12.50, 25.00);
at +14.00 lock +6.00; at +32.00 switch permanently to trail 12.00;
pending expiry 15 min; maximum hold 480 min;
risk-based volume: 3.5% of balance at the initial stop.
4. TOKYO1500
Every weekday at 15:00 Asia/Tokyo.
offset 3.00; initial SL 40.00; activation 3.00;
adaptive trail min(15 * median true range(10), 12.50);
at +10.00 lock +6.00; at +20.00 switch permanently to trail 11.00;
pending expiry 15 min; maximum hold 180 min;
risk-based volume: 0.5% of balance at the initial stop.
5. NYSE1545
Every weekday at 15:45 America/New_York.
offset 2.00; initial SL 40.00;
frozen pre-event activation clip(3 * SMA(true range, 10), 2.50, 5.00);
adaptive trail clip(0.30 * SMA(true range, 10), 0.40, 0.85);
pending expiry 5 min; maximum hold 480 min;
risk-based volume: 2.0% of balance at the initial stop.
Concurrency
-----------
Each session subsystem may have at most one pending OCO pair or one open position.
NEWS is keyed by event minute: every distinct enabled-code timetable minute may have
its own pending OCO pair or open position, even while earlier NEWS positions remain
open. Rows without an enabled code in traded_codes.csv are loaded but not traded.
Simultaneous macro releases must be combined into one timetable row, so one UTC
minute always produces at most one NEWS OCO pair. New setups are also governed by
a 23.5% aggregate portfolio initial-stop-risk cap; each intact OCO pair counts once.
Time-zone and scheduling guarantees
-----------------------------------
* All internal timestamps and timetable timestamps are UTC-aware.
* The host Windows/VPS local time zone is never used for strategy scheduling.
* America/New_York ZoneInfo handles US daylight-saving changes automatically.
* Asia/Tokyo ZoneInfo is used for the two Japanese session events.
* MT5 broker-clock timestamps are converted dynamically to and from true UTC
using the live quote clock; no fixed UTC+2/UTC+3 offset is hard-coded.
* timetable.csv and traded_codes.csv are re-read after edits; temporarily
incomplete/invalid replacements do not replace the last valid in-memory data.
Execution and recovery
----------------------
* Orders are submitted approximately 10 seconds before each event.
* There is no live-spread rejection filter.
* Every pending leg carries a visible broker-side initial stop-loss.
* OCO is client-enforced: after one leg fills, the sibling is removed.
* If both legs fill during a disconnection, the later position is closed after
reconnection; every filled position remains protected by its attached SL.
* Trailing uses the final actual bid/ask tick of each fully completed M1 candle.
Distance is volatility-adaptive under the deployment-safe formulas; the entry
candle is excluded, fixed profit floors/late trails remain, and SL never loosens.
* After restart, trailing state is rebuilt from MT5 tick history. Existing
version-1 state files remain valid; no trailing-state migration is required.
* MT5 network status is checked explicitly. A connection retcode aborts
filling/expiry fallbacks immediately and causes a clean reconnect attempt.
* A hedging account is mandatory.
Installation:
pip install MetaTrader5 tzdata
Run while the intended MT5 terminal/account is open and Algo Trading is enabled:
python xauusd_five_strategy_oco_bot_risk_vector_v2.py
"""
from __future__ import annotations
import csv
import json
import logging
from logging.handlers import RotatingFileHandler
import math
import os
from pathlib import Path
import signal
import sys
import time
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Any, Iterable, Optional
from types import SimpleNamespace
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
try:
import MetaTrader5 as mt5
except ImportError as exc: # pragma: no cover - runtime environment specific
raise SystemExit(
"MetaTrader5 package is not installed. Run: pip install MetaTrader5"
) from exc
# =============================================================================
# USER CONFIGURATION
# =============================================================================
# Exact broker symbol is tried first; common suffix/prefix variants are detected.
PREFERRED_SYMBOL = "XAUUSD"
# -----------------------------------------------------------------------------
# RISK SIZING — MAIN VARIABLES TO ADJUST
# -----------------------------------------------------------------------------
# Values are percentages of the current realized account balance. Volume is
# floored to the broker's lot step so the target risk is never exceeded.
NEWS_RISK_PERCENT = 6.5
NY0830_RISK_PERCENT = 10.5
TOKYO0845_RISK_PERCENT = 3.5
TOKYO1500_RISK_PERCENT = 0.5
NYSE1545_RISK_PERCENT = 2.0
# Aggregate nominal initial-stop risk of all this bot's live positions and OCO
# groups. An intact two-leg OCO pair counts once because only one leg is intended
# to fill. A position plus an unremoved sibling is counted conservatively as both.
PORTFOLIO_RISK_CAP_PERCENT = 23.5
# Conservative allowance used only for position sizing and cap accounting. The
# backtest used $0.15 adverse entry plus $0.15 adverse exit slippage per ounce.
RISK_COST_BUFFER_PER_OUNCE = 0.30
# "SKIP" preserves the requested percentage when the calculated volume is below
# the broker minimum. "FORCE_MIN" trades the broker minimum and may exceed it.
MINIMUM_LOT_POLICY = "SKIP"
# While a setup is blocked by risk, the bot keeps checking until shortly before
# the event. This allows an existing position to close and free risk capacity.
RISK_SKIP_FINALIZE_SECONDS = 0.75
# Optional terminal/account locks. Set these before live deployment when multiple
# terminals or accounts exist on the VPS. None accepts the currently active one.
MT5_TERMINAL_PATH: Optional[str] = None
EXPECTED_ACCOUNT_LOGIN: Optional[int] = None
EXPECTED_SERVER: Optional[str] = None
# Unique IDs. Do not reuse these magic numbers in another EA/script on the account.
NEWS_MAGIC = 271800
NY0830_MAGIC = 271801
TOKYO0845_MAGIC = 271802
TOKYO1500_MAGIC = 271803
NYSE1545_MAGIC = 271804
OUR_MAGICS = {
NEWS_MAGIC,
NY0830_MAGIC,
TOKYO0845_MAGIC,
TOKYO1500_MAGIC,
NYSE1545_MAGIC,
}
# Price both pending legs from the latest live quote approximately ten seconds
# before an event. This leaves operational time for both acknowledgements while
# remaining close to the final pre-event quote used in the M1 backtests.
PLACEMENT_LEAD_SECONDS = 10.0
# Maximum market-order deviation accepted by the client request, in symbol points.
# For a 0.01 XAUUSD point, 15 points = 0.15 USD.
MAX_DEVIATION_POINTS = 15
# Main-loop cadence. It automatically sleeps less when an event is near.
NORMAL_POLL_SECONDS = 0.25
NEAR_EVENT_POLL_SECONDS = 0.05
# Reject stale quotes instead of creating orders from an old market price.
MAX_TICK_AGE_SECONDS = 10.0
# Trailing is allowed to defer locally when the quote feed pauses. This prevents
# an expected short quote gap from aborting reconciliation of the remaining
# subsystems or producing a full top-level traceback on every loop iteration.
TRAIL_TICK_WARNING_INTERVAL_SECONDS = 30.0
# Timetable is reloaded automatically after an edit. The mtime check is cheap
# and a five-second interval keeps last-minute calendar corrections practical.
TIMETABLE_RELOAD_SECONDS = 5
# Runtime connection and history behaviour.
RECONNECT_RETRY_SECONDS = 5
TRADE_LOG_LOOKBACK_DAYS = 45
STATE_RETENTION_DAYS = 120
# Optional hard volume cap. Set to None to use the broker's maximum.
MAX_LOT: Optional[float] = None
# Strategy parameters, in XAUUSD price units (USD per ounce).
NEWS_OFFSET = 2.00
NEWS_INITIAL_SL = 30.00
NEWS_TRAIL_ACTIVATION = 5.00
NEWS_TRAIL_DISTANCE = 0.75
NEWS_TRAIL_ESTIMATOR = "CLOSE_RMS"
NEWS_TRAIL_WINDOW = 5
NEWS_TRAIL_MULTIPLIER = 0.25
NEWS_TRAIL_FLOOR = 0.50
NEWS_TRAIL_CAP = 2.25
NEWS_EXPIRY_SECONDS = 60
NEWS_MAX_HOLD_MINUTES = 120
NEWS_MIN_LOT = 0.01
# NEWS event types are controlled by traded_codes.csv beside this script.
# The bot reloads that file automatically. A timetable minute is tradable when at
# least one of its component event codes is enabled there. The full timetable is
# still loaded; The5ers High Stakes continues to use every enabled timetable row
# for blackout protection, regardless of traded_codes.csv.
TRADED_CODES_RELOAD_SECONDS = 5
SESSION_MIN_LOT = 0.01
NY0830_OFFSET = 6.00
NY0830_INITIAL_SL = 60.00
NY0830_TRAIL_ACTIVATION = 5.00
NY0830_TRAIL_DISTANCE = 6.00
NY0830_TRAIL_ESTIMATOR = "CLOSE_RMS"
NY0830_TRAIL_WINDOW = 20
NY0830_TRAIL_MULTIPLIER = 2.00
NY0830_TRAIL_FLOOR = 3.00
NY0830_TRAIL_CAP = 12.00
NY0830_PROFIT_FLOOR_THRESHOLD = 8.00
NY0830_PROFIT_FLOOR = 5.00
NY0830_EXPIRY_SECONDS = 15 * 60
NY0830_MAX_HOLD_MINUTES = 300
TOKYO0845_OFFSET = 3.00
TOKYO0845_INITIAL_SL = 30.00
TOKYO0845_TRAIL_ACTIVATION = 3.00
TOKYO0845_TRAIL_DISTANCE = 25.00
TOKYO0845_TRAIL_ESTIMATOR = "ATR_WILDER"
TOKYO0845_TRAIL_WINDOW = 30
TOKYO0845_TRAIL_MULTIPLIER = 30.00
TOKYO0845_TRAIL_FLOOR = 12.50
TOKYO0845_TRAIL_CAP = 25.00
TOKYO0845_PROFIT_FLOOR_THRESHOLD = 14.00
TOKYO0845_PROFIT_FLOOR = 6.00
TOKYO0845_LATE_TRAIL_THRESHOLD = 32.00
TOKYO0845_LATE_TRAIL_DISTANCE = 12.00
TOKYO0845_EXPIRY_SECONDS = 15 * 60
TOKYO0845_MAX_HOLD_MINUTES = 480
TOKYO1500_OFFSET = 3.00
TOKYO1500_INITIAL_SL = 40.00
TOKYO1500_TRAIL_ACTIVATION = 3.00
TOKYO1500_TRAIL_DISTANCE = 12.50
TOKYO1500_TRAIL_ESTIMATOR = "TR_MEDIAN"
TOKYO1500_TRAIL_WINDOW = 10
TOKYO1500_TRAIL_MULTIPLIER = 15.00
TOKYO1500_TRAIL_FLOOR = 0.00
TOKYO1500_TRAIL_CAP = 12.50
TOKYO1500_PROFIT_FLOOR_THRESHOLD = 10.00
TOKYO1500_PROFIT_FLOOR = 6.00
TOKYO1500_LATE_TRAIL_THRESHOLD = 20.00
TOKYO1500_LATE_TRAIL_DISTANCE = 11.00
TOKYO1500_EXPIRY_SECONDS = 15 * 60
TOKYO1500_MAX_HOLD_MINUTES = 180
NYSE1545_OFFSET = 2.00
NYSE1545_INITIAL_SL = 40.00
NYSE1545_TRAIL_ACTIVATION = 5.00
NYSE1545_TRAIL_DISTANCE = 0.85
NYSE1545_TRAIL_ESTIMATOR = "TR_SMA"
NYSE1545_TRAIL_WINDOW = 10
NYSE1545_TRAIL_MULTIPLIER = 0.30
NYSE1545_TRAIL_FLOOR = 0.40
NYSE1545_TRAIL_CAP = 0.85
NYSE1545_ACTIVATION_ESTIMATOR = "TR_SMA"
NYSE1545_ACTIVATION_WINDOW = 10
NYSE1545_ACTIVATION_MULTIPLIER = 3.00
NYSE1545_ACTIVATION_FLOOR = 2.50
NYSE1545_ACTIVATION_CAP = 5.00
NYSE1545_EXPIRY_SECONDS = 5 * 60
NYSE1545_MAX_HOLD_MINUTES = 480
# No trade is opened if account currency is not USD, because risk sizing and the
# cost buffer are defined in USD account terms.
REQUIRE_USD_ACCOUNT = True
# =============================================================================
# PATHS AND GLOBALS
# =============================================================================
UTC = timezone.utc
try:
NY_TZ = ZoneInfo("America/New_York")
TOKYO_TZ = ZoneInfo("Asia/Tokyo")
except ZoneInfoNotFoundError as exc: # common on a fresh Windows Python install
raise SystemExit(
"IANA timezone data is missing. Install it with: pip install tzdata"
) from exc
BASE_DIR = Path(__file__).resolve().parent
TIMETABLE_PATH = BASE_DIR / "timetable.csv"
TRADED_CODES_PATH = BASE_DIR / "traded_codes.csv"
LOG_DIR = BASE_DIR / "logs"
STATE_DIR = BASE_DIR / "state"
STATE_PATH = STATE_DIR / "strategy_state.json"
LOCK_PATH = STATE_DIR / "xauusd_combined_oco_bot.lock"
LOG_DIR.mkdir(parents=True, exist_ok=True)
STATE_DIR.mkdir(parents=True, exist_ok=True)
LOGGER = logging.getLogger("xauusd_oco")
STOP_REQUESTED = False
SYMBOL = PREFERRED_SYMBOL
SYMBOL_INFO: Any = None
LOCK_HANDLE: Any = None
LAST_TIMETABLE_RELOAD_MONOTONIC = 0.0
TIMETABLE_SIGNATURE: Optional[tuple[int, int]] = None
TIMETABLE_READY = False
LAST_TRADED_CODES_RELOAD_MONOTONIC = 0.0
TRADED_CODES_SIGNATURE: Optional[tuple[int, int]] = None
TRADED_CODES_READY = False
TRADED_NEWS_CODES: frozenset[str] = frozenset()
NEWS_EVENTS: list["NewsEvent"] = []
LAST_DEAL_SCAN_MONOTONIC = 0.0
LAST_LOGGED_BALANCE_TEXT: Optional[str] = None
LAST_BALANCE_LOG_YEAR: Optional[int] = None
LAST_HORIZON_WARNING_DATE: Optional[str] = None
BROKER_CLOCK_BASIS = "broker_epoch_v2_offset_aware"
BROKER_UTC_OFFSET_SECONDS: Optional[int] = None
MAX_BROKER_UTC_OFFSET_SECONDS = 14 * 60 * 60
COMPLETED_CLOSE_CACHE: dict[int, Optional[tuple[datetime, float, float]]] = {}
M1_RATE_CACHE: dict[int, Optional[list["CompletedM1Bar"]]] = {}
M1_RATE_RETRY_AFTER: dict[int, float] = {}
M1_RATE_SOURCE_WARNINGS: set[int] = set()
VOLATILITY_WARNING_KEYS: set[tuple[str, int]] = set()
RISK_BLOCK_WARNING_KEYS: set[tuple[str, str]] = set()
TRAIL_TICK_WARNING_LAST_MONOTONIC: dict[int, float] = {}
# =============================================================================
# DATA CLASSES
# =============================================================================
@dataclass(frozen=True)
class StrategyConfig:
name: str
magic: int
group_prefix: str
offset: float
initial_sl: float
trail_activation: float
trail_distance: float
expiry_seconds: int
max_hold_minutes: int
risk_percent: float
min_lot: float
profit_floor_threshold: Optional[float] = None
profit_floor: Optional[float] = None
late_trail_threshold: Optional[float] = None
late_trail_distance: Optional[float] = None
trail_estimator: Optional[str] = None
trail_window: int = 0
trail_multiplier: float = 0.0
trail_floor: float = 0.0
trail_cap: float = 0.0
activation_estimator: Optional[str] = None
activation_window: int = 0
activation_multiplier: float = 0.0
activation_floor: float = 0.0
activation_cap: float = 0.0
@dataclass(frozen=True)
class LotSizingResult:
sizing_balance: float
target_risk_usd: float
risk_per_lot_usd: float
raw_lot: float
lot: float
actual_risk_usd: float
NEWS_CONFIG = StrategyConfig(
name="NEWS",
magic=NEWS_MAGIC,
group_prefix="N",
offset=NEWS_OFFSET,
initial_sl=NEWS_INITIAL_SL,
trail_activation=NEWS_TRAIL_ACTIVATION,
trail_distance=NEWS_TRAIL_DISTANCE,
expiry_seconds=NEWS_EXPIRY_SECONDS,
max_hold_minutes=NEWS_MAX_HOLD_MINUTES,
risk_percent=NEWS_RISK_PERCENT,
min_lot=NEWS_MIN_LOT,
trail_estimator=NEWS_TRAIL_ESTIMATOR,
trail_window=NEWS_TRAIL_WINDOW,
trail_multiplier=NEWS_TRAIL_MULTIPLIER,
trail_floor=NEWS_TRAIL_FLOOR,
trail_cap=NEWS_TRAIL_CAP,
)
NY0830_CONFIG = StrategyConfig(
name="NY0830",
magic=NY0830_MAGIC,
group_prefix="Y",
offset=NY0830_OFFSET,
initial_sl=NY0830_INITIAL_SL,
trail_activation=NY0830_TRAIL_ACTIVATION,
trail_distance=NY0830_TRAIL_DISTANCE,
expiry_seconds=NY0830_EXPIRY_SECONDS,
max_hold_minutes=NY0830_MAX_HOLD_MINUTES,
risk_percent=NY0830_RISK_PERCENT,
min_lot=SESSION_MIN_LOT,
profit_floor_threshold=NY0830_PROFIT_FLOOR_THRESHOLD,
profit_floor=NY0830_PROFIT_FLOOR,
trail_estimator=NY0830_TRAIL_ESTIMATOR,
trail_window=NY0830_TRAIL_WINDOW,
trail_multiplier=NY0830_TRAIL_MULTIPLIER,
trail_floor=NY0830_TRAIL_FLOOR,
trail_cap=NY0830_TRAIL_CAP,
)
TOKYO0845_CONFIG = StrategyConfig(
name="TOKYO0845",
magic=TOKYO0845_MAGIC,
group_prefix="A",
offset=TOKYO0845_OFFSET,
initial_sl=TOKYO0845_INITIAL_SL,
trail_activation=TOKYO0845_TRAIL_ACTIVATION,
trail_distance=TOKYO0845_TRAIL_DISTANCE,
expiry_seconds=TOKYO0845_EXPIRY_SECONDS,
max_hold_minutes=TOKYO0845_MAX_HOLD_MINUTES,
risk_percent=TOKYO0845_RISK_PERCENT,
min_lot=SESSION_MIN_LOT,
profit_floor_threshold=TOKYO0845_PROFIT_FLOOR_THRESHOLD,
profit_floor=TOKYO0845_PROFIT_FLOOR,
late_trail_threshold=TOKYO0845_LATE_TRAIL_THRESHOLD,
late_trail_distance=TOKYO0845_LATE_TRAIL_DISTANCE,
trail_estimator=TOKYO0845_TRAIL_ESTIMATOR,
trail_window=TOKYO0845_TRAIL_WINDOW,
trail_multiplier=TOKYO0845_TRAIL_MULTIPLIER,
trail_floor=TOKYO0845_TRAIL_FLOOR,
trail_cap=TOKYO0845_TRAIL_CAP,
)
TOKYO1500_CONFIG = StrategyConfig(
name="TOKYO1500",
magic=TOKYO1500_MAGIC,
group_prefix="T",
offset=TOKYO1500_OFFSET,
initial_sl=TOKYO1500_INITIAL_SL,
trail_activation=TOKYO1500_TRAIL_ACTIVATION,
trail_distance=TOKYO1500_TRAIL_DISTANCE,
expiry_seconds=TOKYO1500_EXPIRY_SECONDS,
max_hold_minutes=TOKYO1500_MAX_HOLD_MINUTES,
risk_percent=TOKYO1500_RISK_PERCENT,
min_lot=SESSION_MIN_LOT,
profit_floor_threshold=TOKYO1500_PROFIT_FLOOR_THRESHOLD,
profit_floor=TOKYO1500_PROFIT_FLOOR,
late_trail_threshold=TOKYO1500_LATE_TRAIL_THRESHOLD,
late_trail_distance=TOKYO1500_LATE_TRAIL_DISTANCE,
trail_estimator=TOKYO1500_TRAIL_ESTIMATOR,
trail_window=TOKYO1500_TRAIL_WINDOW,
trail_multiplier=TOKYO1500_TRAIL_MULTIPLIER,
trail_floor=TOKYO1500_TRAIL_FLOOR,
trail_cap=TOKYO1500_TRAIL_CAP,
)
NYSE1545_CONFIG = StrategyConfig(
name="NYSE1545",
magic=NYSE1545_MAGIC,
group_prefix="C",
offset=NYSE1545_OFFSET,
initial_sl=NYSE1545_INITIAL_SL,
trail_activation=NYSE1545_TRAIL_ACTIVATION,
trail_distance=NYSE1545_TRAIL_DISTANCE,
expiry_seconds=NYSE1545_EXPIRY_SECONDS,
max_hold_minutes=NYSE1545_MAX_HOLD_MINUTES,
risk_percent=NYSE1545_RISK_PERCENT,
min_lot=SESSION_MIN_LOT,
trail_estimator=NYSE1545_TRAIL_ESTIMATOR,
trail_window=NYSE1545_TRAIL_WINDOW,
trail_multiplier=NYSE1545_TRAIL_MULTIPLIER,
trail_floor=NYSE1545_TRAIL_FLOOR,
trail_cap=NYSE1545_TRAIL_CAP,
activation_estimator=NYSE1545_ACTIVATION_ESTIMATOR,
activation_window=NYSE1545_ACTIVATION_WINDOW,
activation_multiplier=NYSE1545_ACTIVATION_MULTIPLIER,
activation_floor=NYSE1545_ACTIVATION_FLOOR,
activation_cap=NYSE1545_ACTIVATION_CAP,
)
ALL_CONFIGS = (
NEWS_CONFIG,
NY0830_CONFIG,
TOKYO0845_CONFIG,
TOKYO1500_CONFIG,
NYSE1545_CONFIG,
)
CONFIG_BY_MAGIC = {config.magic: config for config in ALL_CONFIGS}
SESSION_CONFIGS = (
NY0830_CONFIG,
TOKYO0845_CONFIG,
TOKYO1500_CONFIG,
NYSE1545_CONFIG,
)
CONFIG_BY_MAGIC = {config.magic: config for config in ALL_CONFIGS}
@dataclass(frozen=True)
class DailySessionSpec:
config: StrategyConfig
timezone: ZoneInfo
hour: int
minute: int
event_id_prefix: str
display_name: str
DAILY_SESSION_SPECS = (
DailySessionSpec(
NY0830_CONFIG,
NY_TZ,
8,
30,
"NY0830",
"NY 08:30 ET",
),
DailySessionSpec(
TOKYO0845_CONFIG,
TOKYO_TZ,
8,
45,
"TOKYO0845",
"Tokyo 08:45 JST",
),
DailySessionSpec(
TOKYO1500_CONFIG,
TOKYO_TZ,
15,
0,
"TOKYO1500",
"Tokyo 15:00 JST",
),
DailySessionSpec(
NYSE1545_CONFIG,
NY_TZ,
15,
45,
"NYSE1545",
"NYSE 15:45 ET",
),
)
@dataclass(frozen=True)
class NewsEvent:
event_id: str
event_name: str
when_utc: datetime
enabled: bool
event_codes: tuple[str, ...]
tradable: bool
# =============================================================================
# LOGGING, LOCKING, STATE
# =============================================================================
def setup_logging() -> None:
LOGGER.setLevel(logging.INFO)
LOGGER.handlers.clear()
formatter = logging.Formatter(
"%(asctime)sZ | %(levelname)-8s | %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
formatter.converter = time.gmtime
file_handler = RotatingFileHandler(
LOG_DIR / "runtime.log",
maxBytes=5_000_000,
backupCount=5,
encoding="utf-8",
)
file_handler.setFormatter(formatter)
LOGGER.addHandler(file_handler)
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setFormatter(formatter)
LOGGER.addHandler(console_handler)
def acquire_single_instance_lock() -> None:
"""Hold an OS file lock for the lifetime of the process."""
global LOCK_HANDLE
LOCK_HANDLE = open(LOCK_PATH, "a+b")
try:
if os.name == "nt":
import msvcrt
LOCK_HANDLE.seek(0)
if LOCK_HANDLE.tell() == 0:
LOCK_HANDLE.write(b"0")
LOCK_HANDLE.flush()
LOCK_HANDLE.seek(0)
msvcrt.locking(LOCK_HANDLE.fileno(), msvcrt.LK_NBLCK, 1)
else: # pragma: no cover - production target is Windows
import fcntl
fcntl.flock(LOCK_HANDLE.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
except (OSError, IOError) as exc:
raise SystemExit("Another instance of this bot is already running.") from exc
LOCK_HANDLE.seek(0)
LOCK_HANDLE.truncate()
LOCK_HANDLE.write(str(os.getpid()).encode("ascii"))
LOCK_HANDLE.flush()
def default_state() -> dict[str, Any]:
return {
"version": 1,
"processed_events": {},
"active_groups": {},
"trail": {},
"logged_position_ids": [],
}
def load_state() -> dict[str, Any]:
if not STATE_PATH.exists():
return default_state()
try:
data = json.loads(STATE_PATH.read_text(encoding="utf-8"))
if not isinstance(data, dict):
raise ValueError("state root is not an object")
base = default_state()
for key, value in base.items():
data.setdefault(key, value)
return data
except Exception:
LOGGER.exception("State file is unreadable; backing it up and rebuilding.")
backup = STATE_PATH.with_suffix(f".corrupt.{int(time.time())}.json")
try:
STATE_PATH.replace(backup)
except OSError:
pass
return default_state()
def save_state(state: dict[str, Any]) -> None:
tmp = STATE_PATH.with_suffix(".tmp")
payload = json.dumps(state, indent=2, sort_keys=True, ensure_ascii=False)
tmp.write_text(payload, encoding="utf-8")
os.replace(tmp, STATE_PATH)
def prune_state(state: dict[str, Any], now_utc: datetime) -> None:
cutoff = now_utc - timedelta(days=STATE_RETENTION_DAYS)
processed = state.get("processed_events", {})
for key in list(processed):
stamp = processed[key].get("updated_utc") if isinstance(processed[key], dict) else None
try:
dt = parse_utc(stamp) if stamp else now_utc
except Exception:
dt = now_utc
if dt < cutoff:
del processed[key]
# Trail state for positions that no longer exist is cleared elsewhere.
logged = state.get("logged_position_ids", [])
if len(logged) > 10_000:
state["logged_position_ids"] = logged[-10_000:]
# =============================================================================
# TIME AND CSV
# =============================================================================
def utc_now() -> datetime:
return datetime.now(UTC)
def format_utc(dt: datetime) -> str:
return dt.astimezone(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
def parse_utc(value: str) -> datetime:
value = value.strip()
if value.endswith("Z"):
value = value[:-1] + "+00:00"
dt = datetime.fromisoformat(value)
if dt.tzinfo is None:
raise ValueError(f"UTC datetime lacks timezone: {value!r}")
return dt.astimezone(UTC)
def bool_from_csv(value: str) -> bool:
return value.strip().lower() in {"1", "true", "yes", "y", "on"}
def refresh_news_tradability() -> None:
"""Re-evaluate loaded timetable rows after traded_codes.csv changes."""
global NEWS_EVENTS
if not NEWS_EVENTS:
return
NEWS_EVENTS = [
NewsEvent(
event_id=event.event_id,
event_name=event.event_name,
when_utc=event.when_utc,
enabled=event.enabled,
event_codes=event.event_codes,
tradable=event.enabled and traded_news_minute(event.event_codes),
)
for event in NEWS_EVENTS
]
def load_traded_codes(force: bool = False) -> None:
"""Atomically load the user-maintained traded_codes.csv file.
Required columns are ``event_code`` and ``enabled``. Extra columns such as
``family`` and ``notes`` are ignored. Missing or invalid replacements never
overwrite the last valid in-memory code set. To intentionally disable all
NEWS entries, keep at least one row and set every row's enabled field to 0.
"""
global TRADED_NEWS_CODES, TRADED_CODES_SIGNATURE
global LAST_TRADED_CODES_RELOAD_MONOTONIC, TRADED_CODES_READY
now_mono = time.monotonic()
if (
not force
and now_mono - LAST_TRADED_CODES_RELOAD_MONOTONIC
< TRADED_CODES_RELOAD_SECONDS
):
return
LAST_TRADED_CODES_RELOAD_MONOTONIC = now_mono
if not TRADED_CODES_PATH.exists():
raise FileNotFoundError(f"Missing traded-code file: {TRADED_CODES_PATH}")
last_change_error: Optional[Exception] = None
for attempt in range(1, 4):
before = TRADED_CODES_PATH.stat()
signature = (before.st_mtime_ns, before.st_size)
if not force and TRADED_CODES_SIGNATURE == signature:
return
enabled_codes: set[str] = set()
seen_codes: set[str] = set()
data_rows = 0
try:
with TRADED_CODES_PATH.open(
"r", newline="", encoding="utf-8-sig"
) as handle:
reader = csv.DictReader(handle)
required = {"event_code", "enabled"}
if reader.fieldnames is None or not required.issubset(
set(reader.fieldnames)
):
raise ValueError(
"traded_codes.csv must contain columns: event_code,enabled"
)
for line_no, row in enumerate(reader, start=2):
if not any((value or "").strip() for value in row.values()):
continue
data_rows += 1
event_code = (row.get("event_code") or "").strip().lower()
enabled_text = (row.get("enabled") or "").strip()
if not event_code:
raise ValueError(
f"Blank event_code at traded_codes.csv line {line_no}"
)
if event_code in seen_codes:
raise ValueError(
f"Duplicate event_code {event_code!r} at "
f"traded_codes.csv line {line_no}"
)
normalized_enabled = enabled_text.lower()
valid_true = {"1", "true", "yes", "y", "on"}
valid_false = {"0", "false", "no", "n", "off"}
if normalized_enabled not in valid_true | valid_false:
raise ValueError(
f"Invalid enabled value {enabled_text!r} at "
f"traded_codes.csv line {line_no}"
)
seen_codes.add(event_code)
if normalized_enabled in valid_true:
enabled_codes.add(event_code)
if data_rows == 0:
raise ValueError(
"traded_codes.csv contains no data rows; retain rows with "
"enabled=0 to intentionally disable NEWS trading"
)
after = TRADED_CODES_PATH.stat()
after_signature = (after.st_mtime_ns, after.st_size)
if signature != after_signature:
last_change_error = RuntimeError(
"traded_codes.csv changed while it was being read"
)
time.sleep(0.10)
continue
TRADED_NEWS_CODES = frozenset(enabled_codes)
TRADED_CODES_SIGNATURE = after_signature
TRADED_CODES_READY = True
refresh_news_tradability()
future_tradable = sum(
event.tradable
and event.when_utc > utc_now() - timedelta(minutes=1)
for event in NEWS_EVENTS
)
LOGGER.info(
"Traded-code file loaded atomically: %d rows, %d enabled codes, "
"%d current/future tradable timetable minute(s); %s",
data_rows,
len(TRADED_NEWS_CODES),
future_tradable,
TRADED_CODES_PATH,
)
if not TRADED_NEWS_CODES:
LOGGER.warning(
"traded_codes.csv currently enables no event codes; NEWS "
"entries are disabled while session strategies continue."
)
return
except (FileNotFoundError, OSError) as exc:
last_change_error = exc
if attempt < 3:
time.sleep(0.10)
continue
raise
except Exception:
raise
if last_change_error is not None:
raise last_change_error
raise RuntimeError("Could not load a stable traded_codes.csv snapshot")
def event_codes_from_csv_row(row: dict[str, str]) -> tuple[str, ...]:
"""Extract normalized MT5 event codes from a timetable row.
Newer exporters may provide a dedicated event_codes column. The current EA
stores the same value inside notes as ``event_codes=a|b|c``. Both layouts are
accepted so the filter remains compatible with future exporter revisions.
"""
raw = (row.get("event_codes") or "").strip()
if not raw:
notes = row.get("notes") or ""
marker = "event_codes="
marker_at = notes.lower().find(marker)
if marker_at >= 0:
raw = notes[marker_at + len(marker):].split(";", 1)[0].strip()
normalized = {
code.strip().lower()
for code in raw.replace(",", "|").split("|")
if code.strip()
}
return tuple(sorted(normalized))
def traded_news_minute(event_codes: tuple[str, ...]) -> bool:
"""Return True when any component code is enabled in traded_codes.csv."""
return bool(TRADED_NEWS_CODES.intersection(event_codes))
def load_timetable(force: bool = False) -> None:
"""Load a complete, stable timetable.csv from the bot's own folder.
The MT5 exporter publishes the file atomically, but this loader still verifies
that size and modification time did not change while it was being parsed.
A failed reload raises without replacing NEWS_EVENTS, so the main loop keeps
the last valid in-memory calendar.
"""
global NEWS_EVENTS, TIMETABLE_SIGNATURE, LAST_TIMETABLE_RELOAD_MONOTONIC
global TIMETABLE_READY
now_mono = time.monotonic()
if not force and now_mono - LAST_TIMETABLE_RELOAD_MONOTONIC < TIMETABLE_RELOAD_SECONDS:
return
LAST_TIMETABLE_RELOAD_MONOTONIC = now_mono
if not TIMETABLE_PATH.exists():
raise FileNotFoundError(f"Missing timetable: {TIMETABLE_PATH}")
last_change_error: Optional[Exception] = None
for attempt in range(1, 4):
before = TIMETABLE_PATH.stat()
signature = (before.st_mtime_ns, before.st_size)
if not force and TIMETABLE_SIGNATURE == signature:
return
rows: list[NewsEvent] = []
seen_ids: set[str] = set()
seen_times: set[datetime] = set()
try:
with TIMETABLE_PATH.open("r", newline="", encoding="utf-8-sig") as handle:
reader = csv.DictReader(handle)
required = {"event_id", "event_name", "utc_datetime", "enabled"}
if reader.fieldnames is None or not required.issubset(set(reader.fieldnames)):
raise ValueError(
"timetable.csv must contain columns: "
"event_id,event_name,utc_datetime,enabled"
)
for line_no, row in enumerate(reader, start=2):
# Permit harmless completely blank lines at the end of a CSV.
if not any((value or "").strip() for value in row.values()):
continue
event_id = (row.get("event_id") or "").strip()
event_name = (row.get("event_name") or "").strip()
utc_text = (row.get("utc_datetime") or "").strip()
enabled_text = (row.get("enabled") or "").strip()
event_codes = event_codes_from_csv_row(row)
if not event_id or not event_name or not utc_text:
raise ValueError(
f"Blank event_id/event_name/utc_datetime at timetable line {line_no}"
)
if event_id in seen_ids:
raise ValueError(f"Duplicate event_id {event_id!r} at line {line_no}")
when = parse_utc(utc_text)
enabled = bool_from_csv(enabled_text)
if enabled and when in seen_times:
raise ValueError(
f"Duplicate enabled UTC minute {format_utc(when)}. "
"Combine simultaneous releases into one row."
)
seen_ids.add(event_id)
if enabled:
seen_times.add(when)
rows.append(
NewsEvent(
event_id=event_id,
event_name=event_name,
when_utc=when,
enabled=enabled,
event_codes=event_codes,
tradable=enabled and traded_news_minute(event_codes),
)
)
after = TIMETABLE_PATH.stat()
after_signature = (after.st_mtime_ns, after.st_size)
if signature != after_signature:
last_change_error = RuntimeError(
"timetable.csv changed while it was being read"
)
time.sleep(0.10)
continue
enabled_count = sum(event.enabled for event in rows)
future_enabled = sum(
event.enabled and event.when_utc > utc_now() - timedelta(minutes=1)
for event in rows
)
tradable_count = sum(event.tradable for event in rows)
future_tradable = sum(
event.tradable and event.when_utc > utc_now() - timedelta(minutes=1)
for event in rows
)
missing_code_count = sum(
event.enabled and not event.event_codes for event in rows
)
if not rows:
raise ValueError("timetable.csv contains no event rows")
if enabled_count == 0:
raise ValueError("timetable.csv contains no enabled events")
if future_enabled == 0:
raise ValueError("timetable.csv contains no current or future enabled events")
NEWS_EVENTS = sorted(rows, key=lambda event: event.when_utc)
TIMETABLE_SIGNATURE = after_signature
TIMETABLE_READY = True
LOGGER.info(
"Timetable loaded atomically: %d rows, %d enabled, %d current/future, "
"%d traded-code rows, %d traded-code current/future; %s",
len(NEWS_EVENTS),
enabled_count,
future_enabled,
tradable_count,
future_tradable,
TIMETABLE_PATH,
)
if missing_code_count:
LOGGER.warning(
"%d enabled timetable row(s) have no event_codes and are fail-closed "
"for NEWS trading.",
missing_code_count,
)
if future_tradable == 0:
LOGGER.warning(
"No current or future timetable minute matches the enabled codes in traded_codes.csv; "
"NEWS trading is disabled until a matching row appears."
)
return
except (FileNotFoundError, OSError) as exc:
last_change_error = exc
if attempt < 3:
time.sleep(0.10)
continue
raise
raise RuntimeError(
"Could not obtain a stable timetable.csv after three read attempts"
) from last_change_error
def today_session_event(
now_utc: datetime,
spec: DailySessionSpec,
) -> Optional[tuple[str, datetime, str]]:
"""Build today's event in its market-local calendar, then convert to UTC.
Host/VPS local time is never consulted. New York DST is handled by ZoneInfo;
Tokyo remains fixed at JST through the same timezone-aware mechanism.
"""
local_now = now_utc.astimezone(spec.timezone)
local_day = local_now.date()
if local_day.weekday() >= 5:
return None
local_event = datetime(
local_day.year,
local_day.month,
local_day.day,
spec.hour,
spec.minute,
tzinfo=spec.timezone,
)
event_utc = local_event.astimezone(UTC)
event_id = f"{spec.event_id_prefix}_{local_day:%Y%m%d}"
return event_id, event_utc, f"{spec.display_name} {local_day.isoformat()}"
def all_daily_session_events(
now_utc: datetime,
) -> list[tuple[DailySessionSpec, str, datetime, str]]:
events: list[tuple[DailySessionSpec, str, datetime, str]] = []
for spec in DAILY_SESSION_SPECS:
event = today_session_event(now_utc, spec)
if event is not None:
event_id, event_utc, event_name = event
events.append((spec, event_id, event_utc, event_name))
return events
def seconds_to_nearest_event(now_utc: datetime) -> float:
candidates: list[float] = []
for event in NEWS_EVENTS:
if event.tradable and event.when_utc >= now_utc - timedelta(seconds=5):
candidates.append(abs((event.when_utc - now_utc).total_seconds()))
break
for _spec, _event_id, event_utc, _event_name in all_daily_session_events(now_utc):
candidates.append(abs((event_utc - now_utc).total_seconds()))
return min(candidates) if candidates else 9_999.0
# =============================================================================
# MT5 CONNECTION AND SYMBOL HELPERS
# =============================================================================
def mt5_error_text() -> str:
try:
return repr(mt5.last_error())
except Exception:
return "unknown MT5 error"
def connect_mt5() -> None:
global SYMBOL, SYMBOL_INFO
initialized = (
mt5.initialize(path=MT5_TERMINAL_PATH)
if MT5_TERMINAL_PATH
else mt5.initialize()
)
if not initialized:
raise RuntimeError(f"mt5.initialize() failed: {mt5_error_text()}")
account = mt5.account_info()
terminal = mt5.terminal_info()
if account is None or terminal is None:
raise RuntimeError(f"MT5 account/terminal unavailable: {mt5_error_text()}")
if hasattr(terminal, "connected") and not bool(terminal.connected):
raise RuntimeError("MT5 terminal is open but not connected to the broker server.")
if EXPECTED_ACCOUNT_LOGIN is not None and int(account.login) != int(EXPECTED_ACCOUNT_LOGIN):
raise RuntimeError(
f"Wrong MT5 account: connected to {account.login}, expected {EXPECTED_ACCOUNT_LOGIN}."
)
if EXPECTED_SERVER is not None and str(account.server) != str(EXPECTED_SERVER):
raise RuntimeError(
f"Wrong MT5 server: connected to {account.server!r}, expected {EXPECTED_SERVER!r}."
)
hedging_mode = getattr(mt5, "ACCOUNT_MARGIN_MODE_RETAIL_HEDGING", 2)
if account.margin_mode != hedging_mode:
raise RuntimeError(
"A hedging MT5 account is mandatory. This account is netting/exchange mode, "
"so the five XAUUSD strategy positions cannot be managed independently."
)
if REQUIRE_USD_ACCOUNT and str(account.currency).upper() != "USD":
raise RuntimeError(
f"Account currency is {account.currency}, but lot formulas are defined for USD balance."
)
if not account.trade_allowed or not account.trade_expert:
raise RuntimeError("Trading or expert trading is disabled on the account.")
if hasattr(terminal, "trade_allowed") and not terminal.trade_allowed:
raise RuntimeError("Algo Trading is disabled in the MT5 terminal.")
SYMBOL = resolve_symbol(PREFERRED_SYMBOL)
if not mt5.symbol_select(SYMBOL, True):
raise RuntimeError(f"Could not select {SYMBOL} in Market Watch: {mt5_error_text()}")
SYMBOL_INFO = mt5.symbol_info(SYMBOL)
if SYMBOL_INFO is None:
raise RuntimeError(f"symbol_info({SYMBOL}) failed: {mt5_error_text()}")
LOGGER.info(
"Connected: account=%s server=%s balance=%.2f currency=%s symbol=%s digits=%d point=%g",
account.login,
account.server,
account.balance,
account.currency,
SYMBOL,
SYMBOL_INFO.digits,
SYMBOL_INFO.point,
)
try:
current_tick(require_fresh=False)
except Exception as exc:
LOGGER.warning(
"MT5 broker clock offset is not yet available; it will be detected "
"from the next usable live quote: %s",
exc,
)
def ensure_connection() -> bool:
"""Verify actual broker connectivity, not merely cached terminal objects."""
global SYMBOL_INFO
account = mt5.account_info()
terminal = mt5.terminal_info()
connected = (
account is not None
and terminal is not None
and (not hasattr(terminal, "connected") or bool(terminal.connected))
)
if connected:
info = mt5.symbol_info(SYMBOL)
if info is not None:
SYMBOL_INFO = info
return True
LOGGER.error("MT5 broker connection unavailable; attempting reconnect: %s", mt5_error_text())
try:
mt5.shutdown()
except Exception:
pass
try:
connect_mt5()
return True
except Exception:
LOGGER.exception("MT5 reconnect failed.")
return False
def resolve_symbol(preferred: str) -> str:
exact = mt5.symbol_info(preferred)
if exact is not None:
return preferred
symbols = mt5.symbols_get()
if not symbols:
raise RuntimeError(f"No MT5 symbols available: {mt5_error_text()}")
preferred_upper = preferred.upper()
candidates = [s.name for s in symbols if preferred_upper in s.name.upper()]
if not candidates:
raise RuntimeError(f"Could not find a symbol containing {preferred!r}.")
# Prefer names beginning with XAUUSD and then the shortest broker suffix.
candidates.sort(key=lambda name: (not name.upper().startswith(preferred_upper), len(name), name))
chosen = candidates[0]
LOGGER.warning("Preferred symbol %s not found; using broker symbol %s.", preferred, chosen)
return chosen
def _mt5_object_time_msc(value: Any) -> int:
"""Return an MT5 object's timestamp in the broker's own epoch space.
RoboForex may expose tick/position/deal timestamps shifted by the broker-server
UTC offset. Arithmetic between two MT5 timestamps is exact because the same
offset is present in both values.
"""
raw_msc = int(getattr(value, "time_msc", 0) or 0)
if raw_msc <= 0:
raw_msc = int(float(getattr(value, "time", 0) or 0) * 1000)
return raw_msc
def _broker_clock_datetime(value: Any) -> datetime:
"""Represent an MT5 timestamp as an aware broker-clock datetime.
This is only an internal coordinate for MT5 history queries and completed-M1
comparisons. It is not used for strategy scheduling or presented as true UTC.
"""
return datetime.fromtimestamp(_mt5_object_time_msc(value) / 1000.0, UTC)
def _broker_elapsed_seconds(earlier: Any, later: Any) -> float:
"""Elapsed real seconds between two MT5 objects, independent of timezone."""
return (_mt5_object_time_msc(later) - _mt5_object_time_msc(earlier)) / 1000.0
def _broker_offset_candidate(clock_tick: Any) -> tuple[int, float]:
"""Return the live whole-hour broker offset and quote age in seconds.
Some RoboForex terminals expose MT5 epoch fields in broker-server wall-clock
space rather than true UTC. A live quote supplies both the broker epoch and a
current wall-clock reference, so their difference identifies the active server
offset without hard-coding summer/winter time.
"""
raw_msc = _mt5_object_time_msc(clock_tick)
if raw_msc <= 0:
raise RuntimeError(f"Tick for {SYMBOL} has no usable timestamp")
wall_msc = int(time.time() * 1000)
offset_seconds = int(round((raw_msc - wall_msc) / 3_600_000)) * 3600
quote_age = (wall_msc + offset_seconds * 1000 - raw_msc) / 1000.0
return offset_seconds, quote_age
def broker_utc_offset_seconds(clock_tick: Any = None) -> int:
"""Return the dynamically detected broker-clock offset from true UTC.
The cached value is refreshed from every usable live tick. It is deliberately
constrained to a plausible civil-time range so a weekend-old quote cannot be
mistaken for a multi-day timezone offset.
"""
global BROKER_UTC_OFFSET_SECONDS
tick = clock_tick if clock_tick is not None else mt5.symbol_info_tick(SYMBOL)
if tick is None:
if BROKER_UTC_OFFSET_SECONDS is not None:
return BROKER_UTC_OFFSET_SECONDS
raise RuntimeError(f"Cannot detect {SYMBOL} broker clock: {mt5_error_text()}")
candidate, quote_age = _broker_offset_candidate(tick)
if abs(candidate) > MAX_BROKER_UTC_OFFSET_SECONDS:
if BROKER_UTC_OFFSET_SECONDS is not None:
return BROKER_UTC_OFFSET_SECONDS
raise RuntimeError(
f"Implausible broker UTC offset {candidate / 3600:+.1f}h; "
f"the latest {SYMBOL} quote is probably too old"
)
if BROKER_UTC_OFFSET_SECONDS != candidate:
previous = BROKER_UTC_OFFSET_SECONDS
BROKER_UTC_OFFSET_SECONDS = candidate
COMPLETED_CLOSE_CACHE.clear()
M1_RATE_CACHE.clear()
M1_RATE_RETRY_AFTER.clear()
M1_RATE_SOURCE_WARNINGS.clear()
if previous is None:
LOGGER.info(
"Detected MT5 broker clock offset UTC%+.0f (quote age %.2fs)",
candidate / 3600.0,
quote_age,
)
else:
LOGGER.warning(
"MT5 broker clock offset changed UTC%+.0f -> UTC%+.0f; "
"cleared time-dependent caches.",
previous / 3600.0,
candidate / 3600.0,
)
return candidate
def utc_to_broker_clock(value_utc: datetime, clock_tick: Any = None) -> datetime:
"""Convert a true UTC datetime into the MT5 broker-clock coordinate."""
if value_utc.tzinfo is None:
raise ValueError("UTC-to-broker conversion requires an aware datetime")
offset = broker_utc_offset_seconds(clock_tick)
return value_utc.astimezone(UTC) + timedelta(seconds=offset)
def broker_clock_to_utc(value_broker: datetime, clock_tick: Any = None) -> datetime:
"""Convert an MT5 broker-clock coordinate into true UTC."""
if value_broker.tzinfo is None:
raise ValueError("Broker-to-UTC conversion requires an aware datetime")
offset = broker_utc_offset_seconds(clock_tick)
return (value_broker.astimezone(UTC) - timedelta(seconds=offset)).astimezone(UTC)
def _mt5_object_utc_datetime(value: Any, clock_tick: Any = None) -> datetime:
"""Return an MT5 object's broker-shifted epoch as a true UTC datetime."""
return broker_clock_to_utc(_broker_clock_datetime(value), clock_tick)
class TransientTickError(RuntimeError):
"""Temporary missing/stale quote condition that is safe to retry."""
def current_tick(require_fresh: bool = True) -> Any:
tick = mt5.symbol_info_tick(SYMBOL)
if tick is None or tick.bid <= 0 or tick.ask <= 0:
raise TransientTickError(f"No valid tick for {SYMBOL}: {mt5_error_text()}")
raw_msc = _mt5_object_time_msc(tick)
offset = broker_utc_offset_seconds(tick)
wall_msc = int(time.time() * 1000)
age = (wall_msc + offset * 1000 - raw_msc) / 1000.0
if require_fresh and abs(age) > MAX_TICK_AGE_SECONDS:
direction = "old" if age >= 0 else "ahead of wall clock"
raise TransientTickError(
f"Stale {SYMBOL} tick: {abs(age):.1f} seconds {direction}"
)
return tick
def normalize_price(price: float) -> float:
return round(float(price), int(SYMBOL_INFO.digits))
def volume_decimals(step: float) -> int:
text = f"{step:.10f}".rstrip("0")
return len(text.split(".")[1]) if "." in text else 0
def normalize_volume(raw: float, minimum: float) -> float:
info = SYMBOL_INFO
step = float(info.volume_step)
broker_min = float(info.volume_min)
broker_max = float(info.volume_max)
effective_min = max(minimum, broker_min)
effective_max = min(broker_max, MAX_LOT) if MAX_LOT is not None else broker_max
normalized_min = math.ceil((effective_min - 1e-12) / step) * step
normalized_max = math.floor((effective_max + 1e-12) / step) * step
if normalized_max + 1e-12 < normalized_min:
raise RuntimeError(
f"Volume constraints are impossible: min={normalized_min}, max={normalized_max}, step={step}"
)
policy = str(MINIMUM_LOT_POLICY).strip().upper()
if policy not in {"SKIP", "FORCE_MIN"}:
raise RuntimeError(
f"MINIMUM_LOT_POLICY must be 'SKIP' or 'FORCE_MIN', got {MINIMUM_LOT_POLICY!r}"
)
# Floor to avoid exceeding the requested percentage risk.
floored = math.floor((raw + 1e-12) / step) * step
if floored + 1e-12 < normalized_min:
if policy == "SKIP":
return 0.0
floored = normalized_min
volume = min(floored, normalized_max)
return round(volume, volume_decimals(step))
def symbol_contract_size() -> float:
value = float(getattr(SYMBOL_INFO, "trade_contract_size", 0.0) or 0.0)
if not math.isfinite(value) or value <= 0:
raise RuntimeError(f"Invalid {SYMBOL} contract size: {value!r}")
return value
def nominal_risk_per_lot(config: StrategyConfig) -> float:
return (float(config.initial_sl) + float(RISK_COST_BUFFER_PER_OUNCE)) * symbol_contract_size()
def nominal_risk_for_volume(config: StrategyConfig, volume: float) -> float:
return nominal_risk_per_lot(config) * float(volume)
def lot_for_config(config: StrategyConfig) -> LotSizingResult:
account = mt5.account_info()
if account is None:
raise RuntimeError(f"account_info() failed: {mt5_error_text()}")
sizing_balance = float(account.balance)
if not math.isfinite(sizing_balance) or sizing_balance <= 0:
raise RuntimeError(f"Invalid account balance for sizing: {sizing_balance!r}")
if not math.isfinite(config.risk_percent) or config.risk_percent <= 0:
raise RuntimeError(f"Invalid {config.name} risk percentage: {config.risk_percent!r}")
risk_per_lot = nominal_risk_per_lot(config)
target_risk = sizing_balance * float(config.risk_percent) / 100.0
raw_lot = target_risk / risk_per_lot
lot = normalize_volume(raw_lot, config.min_lot)
actual_risk = nominal_risk_for_volume(config, lot) if lot > 0 else 0.0
return LotSizingResult(
sizing_balance=sizing_balance,
target_risk_usd=target_risk,
risk_per_lot_usd=risk_per_lot,
raw_lot=raw_lot,
lot=lot,
actual_risk_usd=actual_risk,
)
def estimated_sl_loss(
order_type: int,
volume: float,
entry_price: float,
stop_price: float,
) -> Optional[float]:
"""Return broker-calculated loss at the initial SL, when available."""
try:
value = mt5.order_calc_profit(
int(order_type),
SYMBOL,
float(volume),
float(entry_price),
float(stop_price),
)
except Exception:
return None
if value is None:
return None
return abs(min(0.0, float(value)))
def orders_for_magic(magic: int) -> list[Any]:
orders = mt5.orders_get(symbol=SYMBOL)
if orders is None:
return []
return [order for order in orders if int(order.magic) == magic]
def positions_for_magic(magic: int) -> list[Any]:
positions = mt5.positions_get(symbol=SYMBOL)
if positions is None:
return []
return [position for position in positions if int(position.magic) == magic]
def subsystem_busy(config: StrategyConfig) -> bool:
return bool(orders_for_magic(config.magic) or positions_for_magic(config.magic))
def group_id_from_comment(comment: str, config: StrategyConfig) -> Optional[str]:
"""Extract the stable prefix+YYMMDDHHMM event group from an MT5 comment."""
text = str(comment or "").strip()
expected_length = 1 + 10
if len(text) < expected_length:
return None
candidate = text[:expected_length]
if candidate[0] != config.group_prefix or not candidate[1:].isdigit():
return None
return candidate
def orders_for_group(config: StrategyConfig, group_id: str) -> list[Any]:
return [
order
for order in orders_for_magic(config.magic)
if str(getattr(order, "comment", "")).strip().startswith(group_id)
]
def positions_for_group(config: StrategyConfig, group_id: str) -> list[Any]:
return [
position
for position in positions_for_magic(config.magic)
if str(getattr(position, "comment", "")).strip().startswith(group_id)
]
def event_setup_busy(config: StrategyConfig, event_utc: datetime) -> bool:
"""NEWS is busy only for this event minute; sessions remain subsystem-wide."""
if config.name != "NEWS":
return subsystem_busy(config)
group_id = make_group_id(config, event_utc)
return bool(orders_for_group(config, group_id) or positions_for_group(config, group_id))
def portfolio_initial_risk_snapshot() -> tuple[float, list[str]]:
"""Return aggregate nominal initial-stop risk for this bot's live groups.
An intact pending OCO pair counts once. Duplicate positions are summed. If a
position and an unremoved sibling order coexist, both are counted because a
second fill remains possible until reconciliation removes the sibling.
"""
total = 0.0
details: list[str] = []
for config in ALL_CONFIGS:
position_risk: dict[str, float] = {}
order_risk: dict[str, float] = {}
for position in positions_for_magic(config.magic):
group_id = group_id_from_comment(getattr(position, "comment", ""), config)
key = group_id or f"P{config.magic}:{int(position.ticket)}"
risk = nominal_risk_for_volume(config, float(position.volume))
position_risk[key] = position_risk.get(key, 0.0) + risk
for order in orders_for_magic(config.magic):
group_id = group_id_from_comment(getattr(order, "comment", ""), config)
key = group_id or f"O{config.magic}:{int(order.ticket)}"
risk = nominal_risk_for_volume(config, float(order.volume_current))
# Two intact OCO legs represent one intended position, so use max.
order_risk[key] = max(order_risk.get(key, 0.0), risk)
for key in set(position_risk) | set(order_risk):
risk = position_risk.get(key, 0.0) + order_risk.get(key, 0.0)
total += risk
details.append(f"{config.name}:{key}={risk:.2f}")
return total, sorted(details)
def risk_block_log_once(
event_key: str,
reason: str,
message: str,
*args: Any,
) -> None:
key = (event_key, reason)
if key in RISK_BLOCK_WARNING_KEYS:
return
RISK_BLOCK_WARNING_KEYS.add(key)
LOGGER.warning(message, *args)
def finalize_risk_skip_if_due(
state: dict[str, Any],
event_key: str,
event_name: str,
event_utc: datetime,
status: str,
detail: str,
) -> None:
remaining = (event_utc - utc_now()).total_seconds()
if remaining <= RISK_SKIP_FINALIZE_SECONDS:
mark_event(state, event_key, status, event_utc, detail)
LOGGER.warning("Risk sizing skipped %s event %s: %s", status, event_name, detail)
# =============================================================================
# ORDER SENDING
# =============================================================================
def accepted_retcode(retcode: int) -> bool:
accepted = {
getattr(mt5, "TRADE_RETCODE_DONE", 10009),
getattr(mt5, "TRADE_RETCODE_PLACED", 10008),
getattr(mt5, "TRADE_RETCODE_DONE_PARTIAL", 10010),
}
return int(retcode) in accepted
def connection_retcode(retcode: int) -> bool:
return int(retcode) == int(getattr(mt5, "TRADE_RETCODE_CONNECTION", 10031))
def transient_retcode(retcode: int) -> bool:
names = (
"TRADE_RETCODE_REQUOTE",
"TRADE_RETCODE_PRICE_CHANGED",
"TRADE_RETCODE_PRICE_OFF",
"TRADE_RETCODE_TIMEOUT",
"TRADE_RETCODE_CONNECTION",
"TRADE_RETCODE_TOO_MANY_REQUESTS",
"TRADE_RETCODE_LOCKED",
)
values = {getattr(mt5, name, -999999) for name in names}
return int(retcode) in values
def order_result_text(result: Any) -> str:
if result is None:
return f"None, last_error={mt5_error_text()}"
return (
f"retcode={result.retcode} comment={getattr(result, 'comment', '')!r} "
f"order={getattr(result, 'order', 0)} deal={getattr(result, 'deal', 0)} "
f"price={getattr(result, 'price', 0)}"
)
def recover_matching_request(request: dict[str, Any]) -> Any:
"""Recover an order accepted by the server after an ambiguous client timeout.
Retrying an order_send blindly after a timeout can create a duplicate leg. Every
pending leg has a unique magic/comment pair, so inspect live orders and positions
before any retry. A lightweight result object is returned when the request is
already present on the server.
"""
comment = str(request.get("comment", "")).strip()
magic = int(request.get("magic", 0) or 0)
symbol = str(request.get("symbol", SYMBOL))
if not comment or magic <= 0:
return None
request_type = int(request.get("type", -1))
orders = mt5.orders_get(symbol=symbol)
if orders is not None:
for order in orders:
if int(getattr(order, "magic", 0)) != magic:
continue
if str(getattr(order, "comment", "")).strip() != comment:
continue
if request_type >= 0 and int(getattr(order, "type", -2)) != request_type:
continue
LOGGER.warning(
"Recovered already accepted pending order after ambiguous send: "
"ticket=%s comment=%s",
order.ticket,
comment,
)
return SimpleNamespace(
retcode=getattr(mt5, "TRADE_RETCODE_PLACED", 10008),
comment="recovered existing pending order",
order=int(order.ticket),
deal=0,
price=float(getattr(order, "price_open", 0.0)),
)
positions = mt5.positions_get(symbol=symbol)
if positions is not None:
buy_types = {
int(getattr(mt5, "ORDER_TYPE_BUY", 0)),
int(getattr(mt5, "ORDER_TYPE_BUY_LIMIT", 2)),
int(getattr(mt5, "ORDER_TYPE_BUY_STOP", 4)),
int(getattr(mt5, "ORDER_TYPE_BUY_STOP_LIMIT", 6)),
}
expected_position_type = (
int(getattr(mt5, "POSITION_TYPE_BUY", 0))
if request_type in buy_types
else int(getattr(mt5, "POSITION_TYPE_SELL", 1))
)
for position in positions:
if int(getattr(position, "magic", 0)) != magic:
continue
if str(getattr(position, "comment", "")).strip() != comment:
continue
if int(getattr(position, "type", -2)) != expected_position_type:
continue
LOGGER.warning(
"Recovered position filled after ambiguous order send: "
"ticket=%s comment=%s",
position.ticket,
comment,
)
return SimpleNamespace(
retcode=getattr(mt5, "TRADE_RETCODE_DONE", 10009),
comment="recovered filled position",
order=int(position.ticket),
deal=0,
price=float(getattr(position, "price_open", 0.0)),
)
return None
def check_and_send(request: dict[str, Any], retries: int = 1) -> Any:
"""Validate and send one request without duplicating ambiguous fills.
A broker-connection failure is not a filling-policy problem. It aborts this
request immediately, resets the Python/terminal bridge, and lets the main loop
reconnect and retry the complete pair while the placement window remains open.
"""
last_result = None
for attempt in range(retries + 1):
recovered = recover_matching_request(request)
if recovered is not None:
return recovered
check = mt5.order_check(request)
if check is None:
LOGGER.error("order_check returned None: %s", mt5_error_text())
terminal = mt5.terminal_info()
if terminal is None or (
hasattr(terminal, "connected") and not bool(terminal.connected)
):
try:
mt5.shutdown()
except Exception:
pass
return SimpleNamespace(
retcode=getattr(mt5, "TRADE_RETCODE_CONNECTION", 10031),
comment="order_check unavailable: broker connection lost",
order=0,
deal=0,
price=0.0,
)
return None
if int(check.retcode) != 0:
LOGGER.error(
"order_check rejected request: retcode=%s comment=%r request=%s",
check.retcode,
check.comment,
request,
)
if connection_retcode(int(check.retcode)):
try:
mt5.shutdown()
except Exception:
pass
return check
return None
last_result = mt5.order_send(request)
if last_result is not None and accepted_retcode(last_result.retcode):
return last_result
LOGGER.error("order_send failed: %s request=%s", order_result_text(last_result), request)
if last_result is not None and connection_retcode(last_result.retcode):
LOGGER.error(
"Broker connection failure during order submission; "
"aborting policy fallbacks and forcing reconnect."
)
try:
mt5.shutdown()
except Exception:
pass
return last_result
# A timeout can mean acceptance with a lost acknowledgement. Inspect the
# server before retrying; comments/magic numbers make recovery idempotent.
time.sleep(0.10 * (attempt + 1))
recovered = recover_matching_request(request)
if recovered is not None:
return recovered
if (
last_result is not None
and not transient_retcode(last_result.retcode)
) or attempt >= retries:
break
return last_result
def send_pending_request(request: dict[str, Any]) -> Any:
"""Send a pending leg with broker-compatible expiry/filling fallbacks.
Specified server expiry is preferred. If unsupported, the bot uses GTC and
enforces the same expiry locally. Filling policies vary by broker/symbol, so
RETURN, IOC and FOK are tried without ever duplicating a successfully accepted
magic/comment pair.
"""
time_variants: list[dict[str, Any]] = [dict(request)]
if request.get("type_time") == mt5.ORDER_TIME_SPECIFIED:
fallback = dict(request)
fallback["type_time"] = mt5.ORDER_TIME_GTC
fallback.pop("expiration", None)
time_variants.append(fallback)
filling_candidates = [
request.get("type_filling", getattr(mt5, "ORDER_FILLING_RETURN", 2)),
getattr(mt5, "ORDER_FILLING_RETURN", 2),
getattr(mt5, "ORDER_FILLING_IOC", 1),
getattr(mt5, "ORDER_FILLING_FOK", 0),
]
last_result = None
warned_gtc = False
tried: set[tuple[int, int]] = set()
for variant in time_variants:
if variant.get("type_time") == mt5.ORDER_TIME_GTC and not warned_gtc:
LOGGER.warning(
"Specified pending-order expiration is unavailable; trying GTC. "
"The bot will enforce expiry locally."
)
warned_gtc = True
for filling in filling_candidates:
key = (int(variant.get("type_time", -1)), int(filling))
if key in tried:
continue
tried.add(key)
candidate = dict(variant)
candidate["type_filling"] = int(filling)
last_result = check_and_send(candidate, retries=1)
if last_result is not None and accepted_retcode(last_result.retcode):
return last_result
if last_result is not None and connection_retcode(last_result.retcode):
return last_result
return last_result
def remove_pending_order(order: Any, reason: str) -> bool:
request = {
"action": mt5.TRADE_ACTION_REMOVE,
"order": int(order.ticket),
"symbol": SYMBOL,
"magic": int(order.magic),
"comment": reason[:31],
}
result = mt5.order_send(request)
ok = result is not None and accepted_retcode(result.retcode)
if ok:
LOGGER.info("Removed pending order ticket=%s reason=%s", order.ticket, reason)
else:
LOGGER.error(
"Failed to remove pending order ticket=%s: %s",
order.ticket,
order_result_text(result),
)
return ok
def cancel_all_pending(config: StrategyConfig, reason: str) -> None:
for order in orders_for_magic(config.magic):
remove_pending_order(order, reason)
def pair_comment(group_id: str, side: str) -> str:
# Keep below the typical MT5 31-character comment limit.
return f"{group_id}{side}"[:31]
def make_group_id(config: StrategyConfig, event_utc: datetime) -> str:
# N2608071230 / Y2608071230, plus B/S in the order comment.
return f"{config.group_prefix}{event_utc:%y%m%d%H%M}"
def place_oco_pair(
config: StrategyConfig,
event_id: str,
event_name: str,
event_utc: datetime,
state: dict[str, Any],
) -> bool:
group_id = make_group_id(config, event_utc)
if event_setup_busy(config, event_utc):
return False
tick = current_tick(require_fresh=True)
expiry_utc = event_utc + timedelta(seconds=config.expiry_seconds)
buy_price = normalize_price(float(tick.ask) + config.offset)
sell_price = normalize_price(float(tick.bid) - config.offset)
buy_sl = normalize_price(buy_price - config.initial_sl)
sell_sl = normalize_price(sell_price + config.initial_sl)
sizing = lot_for_config(config)
lot = sizing.lot
balance = sizing.sizing_balance
if lot <= 0:
detail = (
f"{config.name} raw lot {sizing.raw_lot:.4f} is below broker/configured "
f"minimum {max(config.min_lot, float(SYMBOL_INFO.volume_min)):.2f}; "
f"policy={MINIMUM_LOT_POLICY}"
)
risk_block_log_once(
event_id,
"minimum_lot",
"%s event %s waiting/skipping: %s",
config.name,
event_name,
detail,
)
finalize_risk_skip_if_due(
state, event_id, event_name, event_utc, "skipped_minimum_lot", detail
)
return False
current_portfolio_risk, risk_details = portfolio_initial_risk_snapshot()
cap_usd = balance * PORTFOLIO_RISK_CAP_PERCENT / 100.0
projected_risk = current_portfolio_risk + sizing.actual_risk_usd
if projected_risk > cap_usd + 1e-9:
detail = (
f"current={current_portfolio_risk:.2f}, proposed={sizing.actual_risk_usd:.2f}, "
f"projected={projected_risk:.2f}, cap={cap_usd:.2f} "
f"({PORTFOLIO_RISK_CAP_PERCENT:.2f}% of balance {balance:.2f})"
)
risk_block_log_once(
event_id,
"portfolio_cap",
"%s event %s blocked by portfolio risk cap: %s | live=%s",
config.name,
event_name,
detail,
"; ".join(risk_details) if risk_details else "none",
)
finalize_risk_skip_if_due(
state, event_id, event_name, event_utc, "skipped_portfolio_risk", detail
)
return False
buy_risk = estimated_sl_loss(mt5.ORDER_TYPE_BUY, lot, buy_price, buy_sl)
sell_risk = estimated_sl_loss(mt5.ORDER_TYPE_SELL, lot, sell_price, sell_sl)
broker_estimated_risk = (
max(value for value in (buy_risk, sell_risk) if value is not None)
if buy_risk is not None or sell_risk is not None
else None
)
LOGGER.info(
"%s risk sizing | target=%.2f%% | balance=%.2f | raw=%.4f | lot=%.2f | "
"nominal initial risk=%.2f (%.2f%%) | portfolio %.2f -> %.2f / %.2f",
config.name,
config.risk_percent,
balance,
sizing.raw_lot,
lot,
sizing.actual_risk_usd,
100.0 * sizing.actual_risk_usd / balance,
current_portfolio_risk,
projected_risk,
cap_usd,
)
if broker_estimated_risk is not None:
LOGGER.info(
"%s broker order_calc_profit initial-SL estimate=%.2f; sizing model with "
"$%.2f/oz cost buffer=%.2f",
config.name,
broker_estimated_risk,
RISK_COST_BUFFER_PER_OUNCE,
sizing.actual_risk_usd,
)
min_distance = max(
int(getattr(SYMBOL_INFO, "trade_stops_level", 0)),
int(getattr(SYMBOL_INFO, "trade_freeze_level", 0)),
) * float(SYMBOL_INFO.point)
if buy_price - float(tick.ask) < min_distance or float(tick.bid) - sell_price < min_distance:
LOGGER.error(
"%s pair invalid: broker minimum pending distance %.5f exceeds configured offset %.5f",
config.name,
min_distance,
config.offset,
)
return False
expiry_broker = utc_to_broker_clock(expiry_utc, tick)
base = {
"action": mt5.TRADE_ACTION_PENDING,
"symbol": SYMBOL,
"volume": lot,
"deviation": MAX_DEVIATION_POINTS,
"magic": config.magic,
"type_time": mt5.ORDER_TIME_SPECIFIED,
"expiration": int(expiry_broker.timestamp()),
"type_filling": mt5.ORDER_FILLING_RETURN,
}
buy_request = {
**base,
"type": mt5.ORDER_TYPE_BUY_STOP,
"price": buy_price,
"sl": buy_sl,
"tp": 0.0,
"comment": pair_comment(group_id, "B"),
}
sell_request = {
**base,
"type": mt5.ORDER_TYPE_SELL_STOP,
"price": sell_price,
"sl": sell_sl,
"tp": 0.0,
"comment": pair_comment(group_id, "S"),
}
buy_result = send_pending_request(buy_request)
if buy_result is None or not accepted_retcode(buy_result.retcode):
return False
sell_result = send_pending_request(sell_request)
if sell_result is None or not accepted_retcode(sell_result.retcode):
LOGGER.critical(
"%s OCO second leg failed after first leg accepted. group=%s first=%s second=%s",
config.name,
group_id,
order_result_text(buy_result),
order_result_text(sell_result),
)
# Cancel the first leg if it is still pending. If it filled during the tiny
# submission interval, it remains protected by its server-side initial SL.
for order in orders_for_group(config, group_id):
remove_pending_order(order, "PAIR_FAIL")
return bool(positions_for_group(config, group_id))
group_record = {
"strategy": config.name,
"event_id": event_id,
"event_name": event_name,
"event_utc": format_utc(event_utc),
"expiry_utc": format_utc(expiry_utc),
"lot": lot,
"risk_percent": config.risk_percent,
"sizing_balance": balance,
"raw_lot": sizing.raw_lot,
"initial_risk_usd": sizing.actual_risk_usd,
"portfolio_risk_before_usd": current_portfolio_risk,
"portfolio_risk_after_usd": projected_risk,
"portfolio_risk_cap_usd": cap_usd,
"reference_bid": float(tick.bid),
"reference_ask": float(tick.ask),
"buy_order": int(getattr(buy_result, "order", 0)),
"sell_order": int(getattr(sell_result, "order", 0)),
}
state["active_groups"][group_id] = group_record
save_state(state)
LOGGER.info(
"%s OCO placed | event=%s | group=%s | lot=%.2f | ref=%.2f/%.2f | "
"buy_stop=%.2f SL=%.2f | sell_stop=%.2f SL=%.2f | expiry=%s",
config.name,
event_name,
group_id,
lot,
tick.bid,
tick.ask,
buy_price,
buy_sl,
sell_price,
sell_sl,
format_utc(expiry_utc),
)
return True
def close_position(position: Any, comment: str) -> bool:
tick = current_tick(require_fresh=True)
position_type_buy = int(position.type) == int(mt5.POSITION_TYPE_BUY)
order_type = mt5.ORDER_TYPE_SELL if position_type_buy else mt5.ORDER_TYPE_BUY
price = float(tick.bid) if position_type_buy else float(tick.ask)
# RoboForex market execution usually accepts IOC; retry supported policies.
filling_candidates = [
getattr(mt5, "ORDER_FILLING_IOC", 1),
getattr(mt5, "ORDER_FILLING_FOK", 0),
getattr(mt5, "ORDER_FILLING_RETURN", 2),
]
tried: set[int] = set()
for filling in filling_candidates:
if filling in tried:
continue
tried.add(filling)
request = {
"action": mt5.TRADE_ACTION_DEAL,
"symbol": SYMBOL,
"position": int(position.ticket),
"volume": float(position.volume),
"type": order_type,
"price": normalize_price(price),
"deviation": MAX_DEVIATION_POINTS,
"magic": int(position.magic),
"comment": comment[:31],
"type_time": mt5.ORDER_TIME_GTC,
"type_filling": filling,
}
result = mt5.order_send(request)
if result is not None and accepted_retcode(result.retcode):
LOGGER.info(
"Closed position ticket=%s strategy=%s reason=%s result=%s",
position.ticket,
CONFIG_BY_MAGIC.get(int(position.magic), NEWS_CONFIG).name,
comment,
order_result_text(result),
)
return True
LOGGER.error(
"Close attempt failed ticket=%s filling=%s: %s",
position.ticket,
filling,
order_result_text(result),
)
time.sleep(0.1)
tick = current_tick(require_fresh=True)
price = float(tick.bid) if position_type_buy else float(tick.ask)
return False
def modify_position_sl(position: Any, new_sl: float) -> bool:
request = {
"action": mt5.TRADE_ACTION_SLTP,
"symbol": SYMBOL,
"position": int(position.ticket),
"sl": normalize_price(new_sl),
"tp": float(position.tp) if float(position.tp) > 0 else 0.0,
"magic": int(position.magic),
"comment": "TRAIL",
}
result = mt5.order_send(request)
ok = result is not None and accepted_retcode(result.retcode)
if ok:
LOGGER.info(
"Position SL modified ticket=%s old_sl=%.2f new_sl=%.2f",
position.ticket,
float(position.sl),
new_sl,
)
else:
LOGGER.error(
"Position SL modification failed ticket=%s new_sl=%.2f: %s",
position.ticket,
new_sl,
order_result_text(result),
)
return ok
def ensure_initial_protection(position: Any, config: StrategyConfig) -> bool:
"""Ensure a recovered/live position never remains without its initial SL.
Pending orders normally install the SL server-side. This guard repairs a
missing or manually widened SL after a restart. If the market is already at
or beyond the intended stop, the position is closed instead of widening risk.
"""
buy = int(position.type) == int(mt5.POSITION_TYPE_BUY)
target = normalize_price(
float(position.price_open) - config.initial_sl
if buy
else float(position.price_open) + config.initial_sl
)
old_sl = float(getattr(position, "sl", 0.0) or 0.0)
point = float(SYMBOL_INFO.point)
protected = (old_sl >= target - point / 2) if buy else (0 < old_sl <= target + point / 2)
if protected:
return True
tick = current_tick(require_fresh=True)
market = float(tick.bid) if buy else float(tick.ask)
if (buy and market <= target) or ((not buy) and market >= target):
LOGGER.critical(
"%s position ticket=%s lacks its required initial SL and market has "
"already crossed it; closing immediately.",
config.name,
position.ticket,
)
close_position(position, f"{config.name}_INITIAL_GAP")
return False
min_distance = max(
int(getattr(SYMBOL_INFO, "trade_stops_level", 0)),
int(getattr(SYMBOL_INFO, "trade_freeze_level", 0)),
) * point
if (buy and target > market - min_distance) or ((not buy) and target < market + min_distance):
LOGGER.critical(
"%s position ticket=%s cannot restore the intended initial SL without "
"widening risk; closing immediately.",
config.name,
position.ticket,
)
close_position(position, f"{config.name}_SL_REPAIR")
return False
LOGGER.warning(
"%s position ticket=%s has missing/wider SL %.2f; restoring %.2f.",
config.name,
position.ticket,
old_sl,
target,
)
return modify_position_sl(position, target)
# =============================================================================
# COMPLETED-M1 TICK CLOSES AND TRAILING
# =============================================================================
def minute_floor(dt: datetime) -> datetime:
return dt.astimezone(UTC).replace(second=0, microsecond=0)
def completed_minute_close(boundary_broker: datetime) -> Optional[tuple[datetime, float, float]]:
"""Return the final bid/ask of the completed broker-clock M1 candle."""
boundary = minute_floor(boundary_broker)
key = int(boundary.timestamp())
if key in COMPLETED_CLOSE_CACHE:
return COMPLETED_CLOSE_CACHE[key]
start = boundary - timedelta(seconds=90)
end = boundary - timedelta(microseconds=1)
ticks = mt5.copy_ticks_range(SYMBOL, start, end, mt5.COPY_TICKS_ALL)
if ticks is None or len(ticks) == 0:
LOGGER.warning(
"No ticks available for completed broker minute ending %s",
boundary.isoformat(),
)
COMPLETED_CLOSE_CACHE.clear()
COMPLETED_CLOSE_CACHE[key] = None
return None
last = ticks[-1]
last_msc = int(last["time_msc"])
completed_start_msc = int((boundary - timedelta(minutes=1)).timestamp() * 1000)
if last_msc < completed_start_msc:
LOGGER.warning(
"No tick belongs to completed broker minute %s; refusing to carry an older quote forward.",
(boundary - timedelta(minutes=1)).isoformat(),
)
COMPLETED_CLOSE_CACHE.clear()
COMPLETED_CLOSE_CACHE[key] = None
return None
bid = float(last["bid"])
ask = float(last["ask"])
if bid <= 0 or ask <= 0:
COMPLETED_CLOSE_CACHE.clear()
COMPLETED_CLOSE_CACHE[key] = None
return None
result = (boundary - timedelta(minutes=1), bid, ask)
COMPLETED_CLOSE_CACHE.clear() # only the current boundary is useful live
COMPLETED_CLOSE_CACHE[key] = result
return result
@dataclass(frozen=True)
class CompletedM1Bar:
minute_ts: int
open: float
high: float
low: float
close: float
def _clamp(value: float, lower: float, upper: float) -> float:
value = max(float(lower), float(value))
if float(upper) > 0:
value = min(float(upper), value)
return value
def completed_m1_rates(boundary_broker: datetime) -> Optional[list[CompletedM1Bar]]:
"""Return completed M1 bid bars ending before a broker-clock boundary.
This RoboForex terminal exposes rate epochs in the same broker-clock coordinate
as live ticks and positions. Querying in that coordinate prevents a UTC+2/+3
server offset from making the newest completed bars appear unavailable.
"""
boundary = minute_floor(boundary_broker)
key = int(boundary.timestamp())
if key in M1_RATE_CACHE:
return M1_RATE_CACHE[key]
if time.monotonic() < M1_RATE_RETRY_AFTER.get(key, 0.0):
return None
start = boundary - timedelta(days=1)
end = boundary - timedelta(microseconds=1)
rates = mt5.copy_rates_range(SYMBOL, mt5.TIMEFRAME_M1, start, end)
if rates is None or len(rates) == 0:
if key not in M1_RATE_SOURCE_WARNINGS:
M1_RATE_SOURCE_WARNINGS.add(key)
LOGGER.warning(
"No completed M1 rates available through broker boundary %s: %s; retrying.",
boundary.isoformat(),
mt5_error_text(),
)
M1_RATE_RETRY_AFTER[key] = time.monotonic() + 1.0
return None
by_minute: dict[int, CompletedM1Bar] = {}
for rate in rates:
minute_ts = int(rate["time"])
if minute_ts >= key:
continue
open_price = float(rate["open"])
high_price = float(rate["high"])
low_price = float(rate["low"])
close_price = float(rate["close"])
if min(open_price, high_price, low_price, close_price) <= 0:
continue
by_minute[minute_ts] = CompletedM1Bar(
minute_ts=minute_ts,
open=open_price,
high=high_price,
low=low_price,
close=close_price,
)
bars = [by_minute[minute] for minute in sorted(by_minute)]
expected_last = key - 60
if not bars or bars[-1].minute_ts < expected_last:
if key not in M1_RATE_SOURCE_WARNINGS:
M1_RATE_SOURCE_WARNINGS.add(key)
LOGGER.warning(
"Latest completed M1 rate is stale at broker boundary %s; latest=%s; retrying.",
boundary.isoformat(),
datetime.fromtimestamp(bars[-1].minute_ts, UTC).isoformat() if bars else None,
)
M1_RATE_RETRY_AFTER[key] = time.monotonic() + 1.0
return None
M1_RATE_CACHE.clear() # live calculations need only the current/frozen boundary
M1_RATE_RETRY_AFTER.clear()
M1_RATE_SOURCE_WARNINGS.discard(key)
M1_RATE_CACHE[key] = bars
return bars
def volatility_from_bars(
bars: list[CompletedM1Bar],
estimator: str,
window: int,
) -> Optional[float]:
"""Calculate the exact estimator definitions used in the brute-force study."""
if window <= 0 or len(bars) < window + 1:
return None
estimator = str(estimator).upper()
closes = [bar.close for bar in bars]
if estimator == "CLOSE_RMS":
changes = [
closes[index] - closes[index - 1]
for index in range(len(closes) - window, len(closes))
]
return math.sqrt(sum(change * change for change in changes) / window)
true_ranges: list[float] = []
for index in range(1, len(bars)):
bar = bars[index]
previous_close = bars[index - 1].close
true_ranges.append(
max(
bar.high - bar.low,
abs(bar.high - previous_close),
abs(bar.low - previous_close),
)
)
if len(true_ranges) < window:
return None
if estimator == "TR_SMA":
values = true_ranges[-window:]
return sum(values) / window
if estimator == "TR_MEDIAN":
values = sorted(true_ranges[-window:])
middle = window // 2
if window % 2:
return values[middle]
return (values[middle - 1] + values[middle]) / 2.0
if estimator == "ATR_WILDER":
# Conventional Wilder ATR seed followed by recursive smoothing. The
# four-day history above makes the result effectively independent of the
# seed for the selected 30-minute window.
atr = sum(true_ranges[:window]) / window
for true_range in true_ranges[window:]:
atr += (true_range - atr) / window
return atr
raise ValueError(f"Unsupported volatility estimator: {estimator!r}")
def volatility_at_boundary(
boundary_broker: datetime,
estimator: str,
window: int,
) -> Optional[float]:
bars = completed_m1_rates(boundary_broker)
if bars is None:
return None
return volatility_from_bars(bars, estimator, window)
def frozen_activation_for_event(
config: StrategyConfig,
event_utc: datetime,
clock_tick: Any = None,
) -> Optional[float]:
if not config.activation_estimator or config.activation_window <= 0:
return None
event_broker = utc_to_broker_clock(event_utc, clock_tick)
volatility = volatility_at_boundary(
event_broker,
config.activation_estimator,
config.activation_window,
)
if volatility is None or not math.isfinite(volatility):
return None
return _clamp(
config.activation_multiplier * volatility,
config.activation_floor,
config.activation_cap,
)
def _group_event_utc(group_id: str, group: Any) -> Optional[datetime]:
if isinstance(group, dict) and group.get("event_utc"):
try:
return parse_utc(str(group["event_utc"]))
except Exception:
pass
try:
return datetime.strptime(group_id[1:11], "%y%m%d%H%M").replace(tzinfo=UTC)
except Exception:
return None
def effective_trail_activation(
position: Any,
config: StrategyConfig,
state: dict[str, Any],
clock_tick: Any,
) -> float:
"""Return fixed activation, or the frozen pre-event NYSE activation."""
if not config.activation_estimator or config.activation_window <= 0:
return config.trail_activation
group_id = group_id_from_comment(getattr(position, "comment", ""), config)
groups = state.setdefault("active_groups", {})
group = groups.get(group_id) if group_id else None
if isinstance(group, dict):
stored = group.get("frozen_trail_activation")
if stored is not None:
try:
value = float(stored)
if math.isfinite(value) and value > 0:
return value
except (TypeError, ValueError):
pass
event_utc = _group_event_utc(group_id, group) if group_id else None
if event_utc is None:
return config.trail_activation
calculated = frozen_activation_for_event(config, event_utc, clock_tick)
if calculated is None:
warning_key = (f"{config.name}_ACTIVATION", int(event_utc.timestamp()))
if warning_key not in VOLATILITY_WARNING_KEYS:
VOLATILITY_WARNING_KEYS.add(warning_key)
LOGGER.warning(
"%s frozen activation unavailable; using fixed fallback %.2f",
config.name,
config.trail_activation,
)
return config.trail_activation
if isinstance(group, dict):
group["frozen_trail_activation"] = calculated
save_state(state)
return calculated
def effective_trail_distance(
config: StrategyConfig,
boundary_broker: datetime,
broker_min_distance: float,
) -> float:
"""Return deployment-safe adaptive distance, never below broker constraints."""
distance = config.trail_distance
if config.trail_estimator and config.trail_window > 0:
volatility = volatility_at_boundary(
boundary_broker,
config.trail_estimator,
config.trail_window,
)
if volatility is not None and math.isfinite(volatility):
distance = _clamp(
config.trail_multiplier * volatility,
config.trail_floor,
config.trail_cap,
)
else:
warning_key = (config.name, int(minute_floor(boundary_broker).timestamp()))
if warning_key not in VOLATILITY_WARNING_KEYS:
VOLATILITY_WARNING_KEYS.add(warning_key)
LOGGER.warning(
"%s volatility unavailable at %s; using fixed fallback trail %.2f",
config.name,
minute_floor(boundary_broker).isoformat(),
config.trail_distance,
)
return max(float(distance), float(broker_min_distance))
def rebuild_trail_state(
position: Any,
config: StrategyConfig,
clock_tick: Any = None,
) -> dict[str, Any]:
"""Reconstruct completed-M1 executable closes after the entry candle.
Position and tick timestamps remain in MT5 broker-clock space. Their common
offset cancels, so reconstruction is independent of Windows/VPS timezone.
A caller-supplied tick keeps one trailing pass on a single clock snapshot.
"""
entry_dt = _broker_clock_datetime(position)
entry_minute = minute_floor(entry_dt)
if clock_tick is None:
clock_tick = current_tick(require_fresh=False)
end_boundary = minute_floor(_broker_clock_datetime(clock_tick))
start = entry_minute
end = end_boundary - timedelta(microseconds=1)
best: Optional[float] = None
last_minute: Optional[datetime] = None
history_query_succeeded = end <= start
if end > start:
ticks = mt5.copy_ticks_range(SYMBOL, start, end, mt5.COPY_TICKS_ALL)
history_query_succeeded = ticks is not None
if ticks is None:
LOGGER.error(
"Could not rebuild trailing history ticket=%s: %s",
position.ticket,
mt5_error_text(),
)
elif len(ticks):
last_by_minute: dict[int, tuple[float, float]] = {}
for tick in ticks:
minute_ts = int(tick["time_msc"] // 60_000) * 60
last_by_minute[minute_ts] = (float(tick["bid"]), float(tick["ask"]))
buy = int(position.type) == int(mt5.POSITION_TYPE_BUY)
# The entry candle is excluded: only minute starts strictly greater
# than the entry candle start are eligible.
for minute_ts in sorted(last_by_minute):
minute_dt = datetime.fromtimestamp(minute_ts, UTC)
if minute_dt <= entry_minute:
continue
bid, ask = last_by_minute[minute_ts]
executable = bid if buy else ask
if executable <= 0:
continue
if best is None:
best = executable
elif buy:
best = max(best, executable)
else:
best = min(best, executable)
last_minute = minute_dt
LOGGER.info(
"Rebuilt trailing state ticket=%s best=%s last_broker_minute=%s",
position.ticket,
best,
last_minute.isoformat() if last_minute else None,
)
rebuilt_through = (
end_boundary - timedelta(minutes=1)
if history_query_succeeded
else None
)
return {
"strategy": config.name,
"clock_basis": BROKER_CLOCK_BASIS,
"best_executable_close": best,
"last_completed_minute": format_utc(last_minute) if last_minute else None,
"rebuilt_through": format_utc(rebuilt_through) if rebuilt_through else None,
}
def theoretical_trailing_sl(
config: StrategyConfig,
buy: bool,
entry_price: float,
best_close: float,
trail_activation: float,
trail_distance: float,
broker_min_distance: float,
) -> Optional[float]:
"""Return the completed-M1 stop under the adaptive deployment-safe rules.
Volatility is measured only from fully completed M1 bars. The current stop is
evaluated before this candidate is applied, and callers still enforce the
never-loosen rule. Existing profit floors and permanent late trails remain
fixed exactly as in the prior specification.
"""
favourable = best_close - entry_price if buy else entry_price - best_close
if favourable + 1e-9 < trail_activation:
return None
active_distance = max(float(trail_distance), float(broker_min_distance))
if (
config.late_trail_threshold is not None
and config.late_trail_distance is not None
and favourable + 1e-9 >= config.late_trail_threshold
):
active_distance = max(
float(config.late_trail_distance),
float(broker_min_distance),
)
candidate = best_close - active_distance if buy else best_close + active_distance
if (
config.profit_floor_threshold is not None
and config.profit_floor is not None
and favourable + 1e-9 >= config.profit_floor_threshold
):
floor_price = entry_price + config.profit_floor if buy else entry_price - config.profit_floor
candidate = max(candidate, floor_price) if buy else min(candidate, floor_price)
return normalize_price(candidate)
def manage_trailing(position: Any, config: StrategyConfig, state: dict[str, Any]) -> None:
ticket = int(position.ticket)
ticket_key = str(ticket)
# A short broker quote pause is expected occasionally. Defer only this
# position's trailing pass; do not abort the rest of the main reconciliation
# loop. Entry placement still uses the same strict freshness check and is not
# relaxed by this handling.
try:
clock_tick = current_tick(require_fresh=True)
except TransientTickError as exc:
now_mono = time.monotonic()
last_warning = TRAIL_TICK_WARNING_LAST_MONOTONIC.get(ticket, -math.inf)
if now_mono - last_warning >= TRAIL_TICK_WARNING_INTERVAL_SECONDS:
TRAIL_TICK_WARNING_LAST_MONOTONIC[ticket] = now_mono
LOGGER.warning(
"%s trailing deferred ticket=%s current_sl=%.2f: %s",
config.name,
ticket,
float(getattr(position, "sl", 0.0) or 0.0),
exc,
)
return
trail_map = state.setdefault("trail", {})
trail_state = trail_map.get(ticket_key)
if (
not isinstance(trail_state, dict)
or trail_state.get("clock_basis") != BROKER_CLOCK_BASIS
):
# Existing version-1 state files remain valid. Only an open position's old
# trail subrecord is rebuilt once under the corrected broker-clock basis.
trail_state = rebuild_trail_state(position, config, clock_tick)
trail_map[ticket_key] = trail_state
save_state(state)
entry_dt = _broker_clock_datetime(position)
now_broker = _broker_clock_datetime(clock_tick)
latest_completed_minute = minute_floor(now_broker) - timedelta(minutes=1)
last_text = trail_state.get("last_completed_minute")
rebuilt_text = trail_state.get("rebuilt_through")
last_minute = parse_utc(last_text) if last_text else None
rebuilt_through = parse_utc(rebuilt_text) if rebuilt_text else None
covered_candidates = [dt for dt in (last_minute, rebuilt_through) if dt is not None]
covered_minute = max(covered_candidates) if covered_candidates else None
if covered_minute is None or latest_completed_minute > covered_minute + timedelta(minutes=1):
trail_state = rebuild_trail_state(position, config, clock_tick)
trail_map[ticket_key] = trail_state
save_state(state)
completed = completed_minute_close(now_broker)
buy = int(position.type) == int(mt5.POSITION_TYPE_BUY)
if completed is not None:
candle_minute, bid_close, ask_close = completed
entry_minute = minute_floor(entry_dt)
last_text = trail_state.get("last_completed_minute")
last_minute = parse_utc(last_text) if last_text else None
# Entry candle is not eligible, and every minute is processed once.
if candle_minute > entry_minute and (last_minute is None or candle_minute > last_minute):
executable_close = bid_close if buy else ask_close
best = trail_state.get("best_executable_close")
best_value = float(best) if best is not None else None
if best_value is None:
best_value = executable_close
elif buy:
best_value = max(best_value, executable_close)
else:
best_value = min(best_value, executable_close)
trail_state["best_executable_close"] = best_value
trail_state["last_completed_minute"] = format_utc(candle_minute)
trail_state["rebuilt_through"] = format_utc(candle_minute)
trail_state["clock_basis"] = BROKER_CLOCK_BASIS
save_state(state)
best_raw = trail_state.get("best_executable_close")
if best_raw is None:
return
best_close = float(best_raw)
entry_price = float(position.price_open)
# Use the same fresh snapshot obtained at the beginning of this trailing pass.
# A second quote request here previously created another avoidable failure point.
tick = clock_tick
current_bid = float(tick.bid)
current_ask = float(tick.ask)
broker_min_distance = max(
int(getattr(SYMBOL_INFO, "trade_stops_level", 0)),
int(getattr(SYMBOL_INFO, "trade_freeze_level", 0)),
) * float(SYMBOL_INFO.point)
trail_activation = effective_trail_activation(
position,
config,
state,
clock_tick,
)
trail_distance = effective_trail_distance(
config,
minute_floor(now_broker),
broker_min_distance,
)
theoretical_sl = theoretical_trailing_sl(
config,
buy,
entry_price,
best_close,
trail_activation,
trail_distance,
broker_min_distance,
)
if theoretical_sl is None:
return
# Conservative restart/gap handling: if the market is already through the
# recovered stop, close at the current market price instead of pretending the
# old stop could be installed retroactively.
if (buy and current_bid <= theoretical_sl) or (not buy and current_ask >= theoretical_sl):
LOGGER.warning(
"Price is through theoretical trailing stop; closing at market. "
"ticket=%s theoretical_sl=%.2f market=%.2f/%.2f",
position.ticket,
theoretical_sl,
current_bid,
current_ask,
)
close_position(position, f"{config.name}_TRAIL_GAP")
return
min_distance = broker_min_distance
if buy:
candidate = min(theoretical_sl, current_bid - min_distance)
candidate = normalize_price(candidate)
old_sl = float(position.sl)
if old_sl > 0 and candidate <= old_sl + float(SYMBOL_INFO.point) / 2:
return
if candidate >= current_bid:
return
else:
candidate = max(theoretical_sl, current_ask + min_distance)
candidate = normalize_price(candidate)
old_sl = float(position.sl)
if old_sl > 0 and candidate >= old_sl - float(SYMBOL_INFO.point) / 2:
return
if candidate <= current_ask:
return
modify_position_sl(position, candidate)
# =============================================================================
# OCO, EXPIRY, POSITION MANAGEMENT
# =============================================================================
def order_expiry_utc(order: Any, state: dict[str, Any]) -> Optional[datetime]:
comment = str(getattr(order, "comment", ""))
group_id = comment[:-1] if comment.endswith(("B", "S")) else comment
group = state.get("active_groups", {}).get(group_id)
if isinstance(group, dict) and group.get("expiry_utc"):
try:
return parse_utc(group["expiry_utc"])
except Exception:
pass
raw = int(getattr(order, "time_expiration", 0) or 0)
if raw > 0:
try:
clock_tick = current_tick(require_fresh=False)
return broker_clock_to_utc(datetime.fromtimestamp(raw, UTC), clock_tick)
except Exception:
LOGGER.exception(
"Could not convert broker expiration for order ticket=%s",
getattr(order, "ticket", 0),
)
return None
def pending_oco_pair_is_complete(orders: list[Any]) -> bool:
"""Return True only for exactly one BUY leg and one SELL leg of one group."""
if len(orders) != 2:
return False
comments = [str(getattr(order, "comment", "")).strip() for order in orders]
if not all(comment.endswith(("B", "S")) for comment in comments):
return False
group_ids = {comment[:-1] for comment in comments}
sides = {comment[-1] for comment in comments}
if len(group_ids) != 1 or sides != {"B", "S"}:
return False
order_types = {int(getattr(order, "type", -1)) for order in orders}
required_types = {
int(getattr(mt5, "ORDER_TYPE_BUY_STOP", 4)),
int(getattr(mt5, "ORDER_TYPE_SELL_STOP", 5)),
}
return order_types == required_types
def manage_open_position(
position: Any,
config: StrategyConfig,
state: dict[str, Any],
) -> None:
if not ensure_initial_protection(position, config):
return
# Maximum hold is elapsed time, not a wall-clock comparison. The live MT5
# tick and the position opening timestamp share the same broker clock, so the
# timezone offset cancels exactly.
clock_tick = current_tick(require_fresh=False)
elapsed_seconds = _broker_elapsed_seconds(position, clock_tick)
if elapsed_seconds < -5.0:
LOGGER.error(
"%s position ticket=%s has a future broker timestamp; deferring time exit. "
"position_msc=%s tick_msc=%s",
config.name,
position.ticket,
_mt5_object_time_msc(position),
_mt5_object_time_msc(clock_tick),
)
elif elapsed_seconds >= config.max_hold_minutes * 60.0:
LOGGER.info(
"%s maximum hold reached ticket=%s elapsed=%.1fs limit=%ss",
config.name,
position.ticket,
elapsed_seconds,
config.max_hold_minutes * 60,
)
close_position(position, f"{config.name}_TIME")
return
manage_trailing(position, config, state)
def reconcile_news_events(config: StrategyConfig, state: dict[str, Any]) -> None:
"""Reconcile each NEWS event minute independently.
Distinct event minutes may have simultaneous positions. OCO sibling removal,
duplicate-fill repair, expiry, time exit and trailing are all scoped to one
group_id, never to every position sharing NEWS_MAGIC.
"""
all_orders = sorted(
orders_for_magic(config.magic),
key=lambda item: (int(getattr(item, "time_setup_msc", 0)), int(item.ticket)),
)
all_positions = sorted(
positions_for_magic(config.magic),
key=lambda item: (int(getattr(item, "time_msc", 0)), int(item.ticket)),
)
grouped_orders: dict[str, list[Any]] = {}
grouped_positions: dict[str, list[Any]] = {}
malformed_orders: list[Any] = []
malformed_positions: list[Any] = []
for order in all_orders:
group_id = group_id_from_comment(getattr(order, "comment", ""), config)
if group_id is None:
malformed_orders.append(order)
else:
grouped_orders.setdefault(group_id, []).append(order)
for position in all_positions:
group_id = group_id_from_comment(getattr(position, "comment", ""), config)
if group_id is None:
malformed_positions.append(position)
else:
grouped_positions.setdefault(group_id, []).append(position)
for order in malformed_orders:
LOGGER.critical(
"NEWS pending order ticket=%s has an unrecognized event comment %r; removing it.",
order.ticket,
str(getattr(order, "comment", "")),
)
remove_pending_order(order, "NEWS_COMMENT_REPAIR")
now = utc_now()
group_ids = set(grouped_orders) | set(grouped_positions)
for group_id in sorted(group_ids):
orders = grouped_orders.get(group_id, [])
positions = grouped_positions.get(group_id, [])
if not positions and orders and not pending_oco_pair_is_complete(orders):
LOGGER.critical(
"NEWS group %s has an incomplete/duplicate OCO set (%d order(s)); "
"cancelling only this event's legs.",
group_id,
len(orders),
)
for order in orders:
remove_pending_order(order, "NEWS_PAIR_REPAIR")
continue
# Local OCO: a fill cancels only the sibling leg from the same event minute.
if positions and orders:
for order in orders:
remove_pending_order(order, "NEWS_OCO")
orders = []
# Both directions may fill during a gap, but only within one event group.
if len(positions) > 1:
keeper = positions[0]
LOGGER.critical(
"NEWS OCO breach for group %s: %d positions. Keeping earliest ticket=%s.",
group_id,
len(positions),
keeper.ticket,
)
for extra in positions[1:]:
close_position(extra, "NEWS_OCO_BREACH")
positions = [keeper]
if not positions:
for order in orders:
expiry = order_expiry_utc(order, state)
if expiry and now >= expiry + timedelta(seconds=1):
remove_pending_order(order, "NEWS_EXPIRED")
continue
manage_open_position(positions[0], config, state)
# A broker may alter a position comment. Do not close such a protected position
# merely because its group cannot be reconstructed; manage it independently.
for position in malformed_positions:
LOGGER.warning(
"NEWS position ticket=%s has unrecognized comment %r; managing it as a "
"standalone NEWS position.",
position.ticket,
str(getattr(position, "comment", "")),
)
manage_open_position(position, config, state)
def reconcile_subsystem(config: StrategyConfig, state: dict[str, Any]) -> None:
if config.name == "NEWS":
reconcile_news_events(config, state)
return
positions = sorted(
positions_for_magic(config.magic),
key=lambda p: (int(getattr(p, "time_msc", 0)), int(p.ticket)),
)
orders = orders_for_magic(config.magic)
# Session subsystems retain the original one-OCO/one-position rule.
if not positions and orders and not pending_oco_pair_is_complete(orders):
LOGGER.critical(
"%s has an incomplete/duplicate pending OCO set (%d order(s)); "
"cancelling all legs before any new placement.",
config.name,
len(orders),
)
for order in orders:
remove_pending_order(order, f"{config.name}_PAIR_REPAIR")
orders = orders_for_magic(config.magic)
if positions and orders:
for order in orders:
remove_pending_order(order, f"{config.name}_OCO")
orders = []
if len(positions) > 1:
keeper = positions[0]
LOGGER.critical(
"%s OCO breach: %d simultaneous positions. Keeping earliest ticket=%s.",
config.name,
len(positions),
keeper.ticket,
)
for extra in positions[1:]:
close_position(extra, f"{config.name}_OCO_BREACH")
positions = [keeper]
now = utc_now()
if not positions:
for order in orders:
expiry = order_expiry_utc(order, state)
if expiry and now >= expiry + timedelta(seconds=1):
remove_pending_order(order, f"{config.name}_EXPIRED")
return
manage_open_position(positions[0], config, state)
def cleanup_trail_state(state: dict[str, Any]) -> None:
live_tickets = {
str(int(position.ticket))
for magic in OUR_MAGICS
for position in positions_for_magic(magic)
}
trail = state.get("trail", {})
changed = False
for ticket in list(trail):
if ticket not in live_tickets:
del trail[ticket]
TRAIL_TICK_WARNING_LAST_MONOTONIC.pop(int(ticket), None)
changed = True
if changed:
save_state(state)
def cleanup_active_groups(state: dict[str, Any]) -> None:
live_comments = {
str(order.comment)
for magic in OUR_MAGICS
for order in orders_for_magic(magic)
} | {
str(position.comment)
for magic in OUR_MAGICS
for position in positions_for_magic(magic)
}
now = utc_now()
groups = state.get("active_groups", {})
changed = False
for group_id, group in list(groups.items()):
still_live = any(comment.startswith(group_id) for comment in live_comments)
if still_live:
continue
try:
expiry = parse_utc(group.get("expiry_utc", ""))
except Exception:
expiry = now - timedelta(days=1)
if now > expiry + timedelta(minutes=10):
del groups[group_id]
changed = True
if changed:
save_state(state)
# =============================================================================
# EVENT PROCESSING
# =============================================================================
def mark_event(
state: dict[str, Any],
event_key: str,
status: str,
event_utc: datetime,
detail: str = "",
) -> None:
state.setdefault("processed_events", {})[event_key] = {
"status": status,
"event_utc": format_utc(event_utc),
"updated_utc": format_utc(utc_now()),
"detail": detail,
}
save_state(state)
def process_scheduled_event(
config: StrategyConfig,
event_key: str,
event_name: str,
event_utc: datetime,
state: dict[str, Any],
) -> None:
if event_key in state.get("processed_events", {}):
return
now = utc_now()
placement_start = event_utc - timedelta(seconds=PLACEMENT_LEAD_SECONDS)
if now < placement_start:
return
# Never create a new pair after the scheduled event time. A late bot start is
# recorded as missed rather than silently changing the tested entry timing.
if now >= event_utc:
status = "skipped_busy" if event_setup_busy(config, event_utc) else "missed_late"
mark_event(state, event_key, status, event_utc, event_name)
LOGGER.warning("%s event %s: %s", config.name, event_name, status)
return
if event_setup_busy(config, event_utc):
# For NEWS this checks only the same UTC event minute. Earlier NEWS event
# positions do not block this event; session subsystems remain single-setup.
return
try:
placed = place_oco_pair(config, event_key, event_name, event_utc, state)
except Exception:
LOGGER.exception("Failed to place %s pair for %s", config.name, event_name)
return
if placed:
RISK_BLOCK_WARNING_KEYS.discard((event_key, "minimum_lot"))
RISK_BLOCK_WARNING_KEYS.discard((event_key, "portfolio_cap"))
mark_event(state, event_key, "placed", event_utc, event_name)
def process_events(state: dict[str, Any]) -> None:
now = utc_now()
# Macro-news timetable. These rows are explicitly UTC and independent of the
# machine, VPS, terminal, broker-server, Moldova or European local timezone.
for event in NEWS_EVENTS:
if not event.tradable:
continue
if event.when_utc < now - timedelta(minutes=1):
if event.event_id not in state.get("processed_events", {}):
mark_event(
state,
event.event_id,
"expired_before_start",
event.when_utc,
event.event_name,
)
continue
if event.when_utc > now + timedelta(minutes=2):
break
process_scheduled_event(
NEWS_CONFIG,
event.event_id,
event.event_name,
event.when_utc,
state,
)
# Four daily market-session events. Each event is constructed in its own
# exchange timezone and converted to UTC before scheduling.
for spec, event_id, event_utc, event_name in all_daily_session_events(now):
process_scheduled_event(
spec.config,
event_id,
event_name,
event_utc,
state,
)
def timetable_horizon_warning(now: datetime) -> None:
global LAST_HORIZON_WARNING_DATE
today_key = now.date().isoformat()
if LAST_HORIZON_WARNING_DATE == today_key:
return
LAST_HORIZON_WARNING_DATE = today_key
future = [event for event in NEWS_EVENTS if event.tradable and event.when_utc > now]
if not future:
LOGGER.critical("No future traded-code NEWS events are available; check traded_codes.csv and timetable.csv.")
return
latest = max(event.when_utc for event in future)
if latest < now + timedelta(days=10):
LOGGER.warning(
"Strong-news timetable horizon is short: last tradable event is %s; check the calendar EA.",
format_utc(latest),
)
# =============================================================================
# BALANCE AND TRADE LOGS
# =============================================================================
def read_last_logged_balance_text(path: Path) -> Optional[str]:
"""Return the last valid balance recorded in *path*, normalized to 2 decimals."""
if not path.exists():
return None
last_balance_text: Optional[str] = None
try:
with path.open("r", encoding="utf-8") as handle:
for raw_line in handle:
line = raw_line.strip()
if not line or "\t" not in line:
continue
_, raw_balance = line.rsplit("\t", 1)
try:
balance = float(raw_balance.strip())
except ValueError:
continue
if math.isfinite(balance):
last_balance_text = f"{balance:.2f}"
except OSError:
LOGGER.exception("Cannot read existing balance log %s", path)
return None
return last_balance_text
def append_balance(force: bool = False) -> None:
"""Append the current balance only when its logged 2-decimal value changes."""
global LAST_LOGGED_BALANCE_TEXT, LAST_BALANCE_LOG_YEAR
now = utc_now()
path = LOG_DIR / f"{now.year}.txt"
# On startup/reconnect calls and at a UTC year rollover, recover the last
# persisted value so an unchanged balance is not duplicated after restart.
if force or LAST_BALANCE_LOG_YEAR != now.year:
LAST_LOGGED_BALANCE_TEXT = read_last_logged_balance_text(path)
LAST_BALANCE_LOG_YEAR = now.year
account = mt5.account_info()
if account is None:
LOGGER.error("Cannot log balance: %s", mt5_error_text())
return
balance_text = f"{float(account.balance):.2f}"
if LAST_LOGGED_BALANCE_TEXT == balance_text:
return
with path.open("a", encoding="utf-8") as handle:
# User requested only date/time and balance, nothing else.
handle.write(f"{now:%Y-%m-%d %H:%M:%S} UTC\t{balance_text}\n")
LAST_LOGGED_BALANCE_TEXT = balance_text
LAST_BALANCE_LOG_YEAR = now.year
def deal_reason_name(value: int) -> str:
mapping = {}
for name in (
"DEAL_REASON_CLIENT",
"DEAL_REASON_MOBILE",
"DEAL_REASON_WEB",
"DEAL_REASON_EXPERT",
"DEAL_REASON_SL",
"DEAL_REASON_TP",
"DEAL_REASON_SO",
"DEAL_REASON_ROLLOVER",
"DEAL_REASON_VMARGIN",
"DEAL_REASON_SPLIT",
"DEAL_REASON_CORPORATE_ACTION",
):
if hasattr(mt5, name):
mapping[int(getattr(mt5, name))] = name.replace("DEAL_REASON_", "")
return mapping.get(int(value), str(value))
def weighted_price(deals: Iterable[Any]) -> float:
deals = list(deals)
total_volume = sum(float(deal.volume) for deal in deals)
if total_volume <= 0:
return 0.0
return sum(float(deal.price) * float(deal.volume) for deal in deals) / total_volume
def append_trade_row(
exit_dt: datetime,
entry_dt: datetime,
strategy: str,
direction: str,
volume: float,
entry_price: float,
exit_price: float,
net: float,
reason: str,
position_id: int,
comments: str,
) -> None:
# Closed trades are grouped by their UTC exit year. Entry and exit fields
# below retain the full UTC date and time for every position.
path = LOG_DIR / f"{exit_dt:%Y}_trades.log"
header = (
"ENTRY DATE/TIME UTC | EXIT DATE/TIME UTC | STRATEGY | DIR | LOT | "
"ENTRY | EXIT | NET USD | OUTCOME | REASON | POSITION | COMMENTS\n"
"--------------------+---------------------+-------------+------+--------+"
"------------+------------+------------+---------+--------------+----------+---------\n"
)
if not path.exists() or path.stat().st_size == 0:
path.write_text(header, encoding="utf-8")
outcome = "PROFIT" if net > 1e-9 else "LOSS" if net < -1e-9 else "FLAT"
line = (
f"{entry_dt:%Y-%m-%d %H:%M:%S} | {exit_dt:%Y-%m-%d %H:%M:%S} | "
f"{strategy:<11} | {direction:<4} | {volume:>6.2f} | "
f"{entry_price:>10.2f} | {exit_price:>10.2f} | {net:>10.2f} | "
f"{outcome:<7} | {reason:<12} | {position_id:<8} | {comments}\n"
)
with path.open("a", encoding="utf-8") as handle:
handle.write(line)
def reconcile_trade_logs(state: dict[str, Any], force: bool = False) -> None:
global LAST_DEAL_SCAN_MONOTONIC
now_mono = time.monotonic()
if not force and now_mono - LAST_DEAL_SCAN_MONOTONIC < 30:
return
LAST_DEAL_SCAN_MONOTONIC = now_mono
end_utc = utc_now() + timedelta(minutes=1)
start_utc = end_utc - timedelta(days=TRADE_LOG_LOOKBACK_DAYS)
try:
clock_tick = current_tick(require_fresh=False)
start_broker = utc_to_broker_clock(start_utc, clock_tick)
end_broker = utc_to_broker_clock(end_utc, clock_tick)
except Exception:
LOGGER.exception("Cannot establish broker-clock range for deal-history reconciliation.")
return
deals = mt5.history_deals_get(start_broker, end_broker)
if deals is None:
LOGGER.error("history_deals_get failed: %s", mt5_error_text())
return
grouped: dict[int, list[Any]] = {}
for deal in deals:
position_id = int(getattr(deal, "position_id", 0) or 0)
if position_id <= 0:
continue
grouped.setdefault(position_id, []).append(deal)
logged = {int(value) for value in state.get("logged_position_ids", [])}
changed = False
entry_in = int(getattr(mt5, "DEAL_ENTRY_IN", 0))
entry_out = int(getattr(mt5, "DEAL_ENTRY_OUT", 1))
entry_out_by = int(getattr(mt5, "DEAL_ENTRY_OUT_BY", 3))
entry_inout = int(getattr(mt5, "DEAL_ENTRY_INOUT", 2))
for position_id, group in sorted(grouped.items()):
if position_id in logged:
continue
if not any(int(getattr(deal, "magic", 0)) in OUR_MAGICS for deal in group):
continue
group.sort(key=lambda deal: (int(getattr(deal, "time_msc", 0)), int(deal.ticket)))
entries = [deal for deal in group if int(deal.entry) in {entry_in, entry_inout}]
exits = [deal for deal in group if int(deal.entry) in {entry_out, entry_out_by, entry_inout}]
if not entries or not exits:
continue
entry_volume = sum(float(deal.volume) for deal in entries)
exit_volume = sum(float(deal.volume) for deal in exits)
if exit_volume + 1e-9 < entry_volume:
continue # still partially open
entry = entries[0]
magic = next(
(int(deal.magic) for deal in entries if int(deal.magic) in OUR_MAGICS),
int(entry.magic),
)
config = CONFIG_BY_MAGIC.get(magic)
if config is None:
continue
direction = (
"BUY"
if int(entry.type) == int(getattr(mt5, "DEAL_TYPE_BUY", 0))
else "SELL"
)
entry_dt = _mt5_object_utc_datetime(entry, clock_tick)
exit_dt = _mt5_object_utc_datetime(exits[-1], clock_tick)
entry_price = weighted_price(entries)
exit_price = weighted_price(exits)
net = sum(
float(getattr(deal, "profit", 0.0))
+ float(getattr(deal, "commission", 0.0))
+ float(getattr(deal, "swap", 0.0))
+ float(getattr(deal, "fee", 0.0))
for deal in group
)
reasons = sorted({deal_reason_name(int(deal.reason)) for deal in exits})
comments = " / ".join(
dict.fromkeys(str(deal.comment).strip() for deal in group if str(deal.comment).strip())
)
append_trade_row(
exit_dt=exit_dt,
entry_dt=entry_dt,
strategy=config.name,
direction=direction,
volume=entry_volume,
entry_price=entry_price,
exit_price=exit_price,
net=net,
reason=",".join(reasons),
position_id=position_id,
comments=comments,
)
LOGGER.info(
"Trade logged position_id=%s strategy=%s direction=%s net=%.2f",
position_id,
config.name,
direction,
net,
)
logged.add(position_id)
changed = True
if changed:
state["logged_position_ids"] = sorted(logged)
save_state(state)
# =============================================================================
# STARTUP VALIDATION AND MAIN LOOP
# =============================================================================
def describe_configuration() -> None:
LOGGER.info(
"Scheduling is UTC-aware and host-timezone independent | "
"news=timetable UTC | New York=%s | Tokyo=%s | placement lead=%.1fs",
getattr(NY_TZ, "key", "America/New_York"),
getattr(TOKYO_TZ, "key", "Asia/Tokyo"),
PLACEMENT_LEAD_SECONDS,
)
LOGGER.info(
"Risk sizing: portfolio cap=%.2f%% | cost buffer=$%.2f/oz | minimum-lot policy=%s",
PORTFOLIO_RISK_CAP_PERCENT,
RISK_COST_BUFFER_PER_OUNCE,
MINIMUM_LOT_POLICY,
)
for config in ALL_CONFIGS:
trail_rule = (
f"clip({config.trail_multiplier:g}*{config.trail_estimator}{config.trail_window},"
f"{config.trail_floor:g},{config.trail_cap:g})"
if config.trail_estimator and config.trail_window > 0
else f"fixed {config.trail_distance:g}"
)
activation_rule = (
f"frozen clip({config.activation_multiplier:g}*{config.activation_estimator}"
f"{config.activation_window},{config.activation_floor:g},{config.activation_cap:g})"
if config.activation_estimator and config.activation_window > 0
else f"fixed {config.trail_activation:g}"
)
LOGGER.info(
"%s parameters: offset=%.2f SL=%.2f activation=%s trail=%s "
"floor_at=%s floor=%s late_at=%s late_trail=%s expiry=%ss hold=%sm "
"risk=%.2f%% min_lot=%.2f",
config.name,
config.offset,
config.initial_sl,
activation_rule,
trail_rule,
config.profit_floor_threshold,
config.profit_floor,
config.late_trail_threshold,
config.late_trail_distance,
config.expiry_seconds,
config.max_hold_minutes,
config.risk_percent,
config.min_lot,
)
def signal_handler(signum: int, _frame: Any) -> None:
global STOP_REQUESTED
STOP_REQUESTED = True
LOGGER.warning(
"Shutdown requested by signal %s. Existing pending orders/positions for all strategies are left "
"server-side with their initial/current SL and pending expiration.",
signum,
)
def main() -> int:
setup_logging()
acquire_single_instance_lock()
signal.signal(signal.SIGINT, signal_handler)
if hasattr(signal, "SIGTERM"):
signal.signal(signal.SIGTERM, signal_handler)
state = load_state()
try:
try:
load_traded_codes(force=True)
except FileNotFoundError:
LOGGER.error(
"traded_codes.csv is not present; NEWS remains disabled until a "
"valid file appears. Session strategies continue."
)
except Exception:
LOGGER.exception(
"Initial traded-code load failed; NEWS remains disabled until a "
"valid replacement is loaded. Session strategies continue."
)
try:
load_timetable(force=True)
except FileNotFoundError:
LOGGER.error(
"timetable.csv is not present yet; NEWS remains disabled until the "
"EA publishes a valid file. Session strategies continue."
)
except Exception:
LOGGER.exception(
"Initial timetable load failed; NEWS remains disabled until a valid "
"replacement is loaded. Session strategies continue."
)
connect_mt5()
describe_configuration()
append_balance(force=True)
reconcile_trade_logs(state, force=True)
prune_state(state, utc_now())
save_state(state)
while not STOP_REQUESTED:
loop_started = time.monotonic()
try:
if not ensure_connection():
time.sleep(RECONNECT_RETRY_SECONDS)
continue
try:
load_traded_codes(force=False)
except FileNotFoundError:
LOGGER.error(
"traded_codes.csv is temporarily missing; retaining the last "
"valid in-memory code set. Set enabled=0 in a valid file to "
"intentionally disable NEWS trading."
)
except Exception:
LOGGER.exception(
"Traded-code reload failed; retaining the last valid in-memory "
"code set."
)
try:
load_timetable(force=False)
except FileNotFoundError:
LOGGER.error(
"timetable.csv is temporarily missing; retaining the last "
"valid in-memory NEWS calendar while session strategies continue."
)
except Exception:
LOGGER.exception(
"Timetable reload failed; retaining the last valid in-memory "
"NEWS calendar while session strategies continue."
)
now = utc_now()
timetable_horizon_warning(now)
# OCO cancellation and position protection run before new entries.
for config in ALL_CONFIGS:
reconcile_subsystem(config, state)
cleanup_trail_state(state)
cleanup_active_groups(state)
process_events(state)
append_balance(force=False)
reconcile_trade_logs(state, force=False)
except Exception:
LOGGER.exception("Unhandled loop error; bot will continue.")
near_window = max(15.0, PLACEMENT_LEAD_SECONDS + 2.0)
near = seconds_to_nearest_event(utc_now()) <= near_window
target_sleep = NEAR_EVENT_POLL_SECONDS if near else NORMAL_POLL_SECONDS
elapsed = time.monotonic() - loop_started
time.sleep(max(0.01, target_sleep - elapsed))
finally:
try:
reconcile_trade_logs(state, force=True)
except Exception:
LOGGER.exception("Final trade-log reconciliation failed.")
try:
mt5.shutdown()
except Exception:
pass
LOGGER.info("Bot stopped.")
return 0
if __name__ == "__main__":
raise SystemExit(main())
Reference in a new issue View git blame Copy permalink