my_metatrader5_mcp/mtlib/cli.py

"""mtdata-cli 封装 - 所有 MetaTrader5 实时数据/交易调用的唯一出口。

把踩过的坑全部收口在这里, 调用方不允许再写裸 subprocess:
- 交易参数全名: --order-type BUY|SELL (大写), --stop-loss/--take-profit (非 --sl/--tp)
- --dry-run 默认 True (预览不发单), 真实下单显式 dry_run=False
- trade_get_open 返回字段是 items 不是 positions
- side 取值不统一 (buy/sell/BUY/SELL/long/short) -> 归一化
- ticket/order_id 字段名不统一 -> _find_in_response 兜底
- 滚动窗口恒长 -> 用时间戳去重, 禁止用 len()
"""

import json
import subprocess

import pandas as pd

from .data import DATA_COLS

CLI_TIMEOUT = 60


# ---------------- 底层执行 ----------------

def run_cli(args, timeout=CLI_TIMEOUT):
    """执行 mtdata-cli, 返回解析后的 dict。失败返回 {'success': False, ...}。"""
    cmd = ["mtdata-cli", "--json"] + args
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
    except subprocess.TimeoutExpired:
        return {"success": False, "error": "CLI timeout"}
    try:
        return json.loads(result.stdout)
    except (ValueError, json.JSONDecodeError):
        return {"success": False, "error": result.stdout or result.stderr}


def _find_in_response(resp, *keys):
    """在响应里递归一层找字段值 (ticket/order_id/deal/position 命名不统一)。"""
    if not isinstance(resp, dict):
        return None
    for k in keys:
        v = resp.get(k)
        if v not in (None, ''):
            return v
    for payload in (resp.get('order'), resp.get('request'),
                    resp.get('result'), resp.get('details')):
        if isinstance(payload, dict):
            for k in keys:
                v = payload.get(k)
                if v not in (None, ''):
                    return v
    return None


def _normalize_side(raw_side):
    """统一 side 取值 (buy/sell, BUY/SELL, long/short, 1/-1) -> 1/-1/0。"""
    s = str(raw_side).lower()
    if s in ("buy", "long", "1", "+1"):
        return 1
    if s in ("sell", "short", "-1"):
        return -1
    return 0


# ---------------- 数据 ----------------

def fetch_candles(symbol, timeframe='M1', limit=100, include_incomplete=False):
    """拉K线。注意: 返回固定 N 根滚动窗口, 必须按时间戳判断新K线。"""
    with_indicators = ("rsi(14),macd(12,26,9),ema(20),ema(50),atr(14),"
                       "bbands(20,2),stoch(14,3,3),obv,cmf(20),adx(14)")
    args = ["data_fetch_candles", symbol, "--timeframe", timeframe,
            "--limit", str(limit), "--indicators", with_indicators]
    if include_incomplete:
        args.append("--include_incomplete=true")
    return run_cli(args)


def df_from_candles(payload):
    """把 fetch_candles 的返回体转成与分析同构的 DataFrame (实时窗口用)。

    与 mtlib.data.load_data 共用列规范化; 供策略的 generate_signal_at 调用。
    """
    df = pd.DataFrame(payload)
    if 'time' in df.columns:
        df['time'] = pd.to_datetime(df['time'])
        df = df.sort_values('time').reset_index(drop=True)
    for col in DATA_COLS:
        if col in df.columns:
            df[col] = pd.to_numeric(df[col], errors='coerce')
    return df


# ---------------- 账户 / 持仓 ----------------

def get_account_info():
    return run_cli(["trade_account_info"])


def get_open_positions(symbol=None):
    """返回规范结构 {'success': bool, 'items': [...]}。

    注意 CLI 字段是 items, 不是 positions。
    """
    args = ["trade_get_open"]
    if symbol:
        args.append(symbol)
    raw = run_cli(args)
    if not raw.get("success"):
        return raw
    items = raw.get("items") or []
    return {"success": True, "items": items}


def find_position_items(symbol, magic=None, ticket=None):
    """按 symbol(+可选 magic/ticket) 过滤持仓列表。"""
    items = get_open_positions(symbol).get("items", [])
    result = []
    for p in items:
        if str(p.get('symbol', '')).upper() != symbol.upper():
            continue
        if magic is not None and str(p.get('magic', '')) != str(magic):
            continue
        if ticket is not None and str(p.get('ticket', '')) != str(ticket):
            continue
        result.append(p)
    return result


def open_tickets(symbol):
    """当前持仓 ticket 集合 (用于对账)。"""
    return {str(p.get('ticket')) for p in find_position_items(symbol)}


# ---------------- 下单 / 改单 / 平仓 ----------------

def position_sizing(balance, risk_per_trade, sl_dist, contract_size, min_volume=0.01,
                    quote_to_account_rate=1.0):
    """手数计算 - 回测与实盘共用同一口径。

    volume = max(min_volume, round(risk_amount / (sl_dist*contract*rate), 2))
    rate=quote_to_account_rate: 1 单位报价货币 = rate 账户货币 (=1 时即旧公式,
    与 USD 报价对子历史结果恒等; USDJPY 传 1/当时报价, 见 instruments)。
    """
    if balance <= 0 or sl_dist <= 0 or contract_size <= 0 or quote_to_account_rate <= 0:
        return min_volume
    risk_amount = balance * risk_per_trade
    return max(min_volume, round(risk_amount / (sl_dist * contract_size * quote_to_account_rate), 2))


def place_trade(symbol, order_type, volume, stop_loss, take_profit,
                magic, comment, idempotency_key, dry_run=True):
    """下单(市价)。dry_run 默认 True = 预览校验, False = 真实发单。"""
    args = ["trade_place", symbol,
            "--order-type", order_type,     # 必须大写 BUY/SELL
            "--volume", str(volume),
            "--stop-loss", str(stop_loss),  # 全名, 不是 --sl
            "--take-profit", str(take_profit),
            "--magic", str(magic) if magic else "",
            "--comment", comment,
            "--idempotency-key", idempotency_key]
    args += ["--dry-run", "false"] if not dry_run else ["--dry-run", "true"]
    return run_cli(args)


def modify_sl(ticket, new_sl, dry_run=True, digits=2):
    """修改止损。

    digits: 报价小数位 (XAUUSD=2, EURUSD=5)。旧版硬编码 2 位会破坏 5 位报价
    的移动止损 (如 1.15420 -> 1.15), 现由调用方传 PARAMS.info.price_digits。
    """
    args = ["trade_modify", "--ticket", str(ticket),
            "--stop-loss", str(round(float(new_sl), digits))]
    args += ["--dry-run", "false"] if not dry_run else ["--dry-run", "true"]
    return run_cli(args)


def close_position(ticket, dry_run=True):
    """市价平仓。"""
    args = ["trade_close", "--ticket", str(ticket)]
    args += ["--dry-run", "false"] if not dry_run else ["--dry-run", "true"]
    return run_cli(args)


def extract_ticket(response):
    """从下单响应里取 ticket (字段名不统一, 用兜底查找)。"""
    return _find_in_response(response, 'ticket', 'order_id', 'deal', 'position')