#!/usr/bin/env python3 """ Скрипт для автоматического создания торговых ордеров на основе подготовленных параметров. Использование: python src/create_trade_orders.py --ticker SBER --direction BUY --entry 100.0 --sl 95.0 --tp 110.0 --risk 1.0 python src/create_trade_orders.py --ticker SBER --direction BUY --entry 100.0 --sl 95.0 --tp 110.0 --risk 1.0 --confidence 80 Функции: - Проверка баланса счёта - Расчёт размера позиции - Создание ордеров: вход, стоп-лосс, тейк-профит - Логирование сделки в журнале - Валидация параметров сделки """ import argparse import json import os import sys from datetime import datetime # Добавляем корень проекта в sys.path PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0, PROJECT_ROOT) from src.api.tbank import ( set_tbank_env, get_account_balance, get_figi, calculate_position_size, place_limit_order, place_stop_order, round_to_tick, get_lot_size ) from src.api.trade_journal import journal from src.utils.config import config def format_currency(amount: float, currency: str = "RUB") -> str: """Форматирование суммы с валютой.""" if currency == "RUB": return f"{amount:,.2f} ₽" else: return f"{amount:,.2f} {currency}" def check_balance(account_id: str = None) -> dict: """Проверить баланс счёта и вернуть информацию.""" print("🏦 Проверка баланса счёта...") try: balance_info = get_account_balance(account_id) if balance_info.get("error"): print(f"❌ Ошибка получения баланса: {balance_info.get('error')}") return None # Общая информация total_amount = balance_info.get('total_amount', 0) total_currency = balance_info.get('total_amount_currency', 'RUB') status = balance_info.get('status', 'N/A') print(f" Общий баланс: {format_currency(total_amount, total_currency)}") print(f" Валюта баланса: {total_currency}") print(f" Статус: {status}") # Обрабатываем валютные балансы currencies = balance_info.get("currencies", []) total_rub = 0.0 if currencies: print(f"\n💳 Валютные балансы:") for currency in currencies: curr_type = currency.get("currency", "N/A") available = float(currency.get("available", 0)) blocked = float(currency.get("blocked", 0)) balance = float(currency.get("balance", 0)) if available > 0 or balance > 0: rub_value = available # Упрощённо - в рублях total_rub += rub_value print(f" {curr_type:4}: {format_currency(available, curr_type)} свободно " f"({format_currency(blocked, curr_type)} заблокировано, " f"всего: {format_currency(balance, curr_type)})") print(f"\n📊 Итог:") print(f" Общий свободный баланс: {format_currency(total_rub, 'RUB')}") return { "available_balance": total_rub, "total_amount": total_amount, "total_currency": total_currency, "currencies": currencies, "status": status } except Exception as e: print(f"❌ Ошибка при получении баланса: {e}") return None def validate_trade_parameters(ticker: str, direction: str, entry: float, sl: float, tp: float) -> tuple[bool, list[str]]: """Валидация параметров сделки.""" errors = [] # Проверка тикера try: figi = get_figi(ticker) except ValueError: errors.append(f"FIGI не найден для тикера {ticker}") return False, errors # Проверка направления direction = direction.upper() if direction not in ("BUY", "SELL"): errors.append(f"Некорректное направление: {direction}. Допустимо: BUY, SELL") # Проверка цен if direction == "BUY": if not (sl < entry < tp): errors.append(f"BUY: должно быть SL ({sl}) < Entry ({entry}) < TP ({tp})") elif direction == "SELL": if not (tp < entry < sl): errors.append(f"SELL: должно быть TP ({tp}) < Entry ({entry}) < SL ({sl})") if entry <= 0 or sl <= 0 or tp <= 0: errors.append("Цены должны быть положительными") # Проверка шага цены и лота try: entry_tick = round_to_tick(entry) sl_tick = round_to_tick(sl) tp_tick = round_to_tick(tp) if entry != entry_tick: print(f" Входная цена округлена: {entry} → {entry_tick}") if sl != sl_tick: print(f" SL округлен: {sl} → {sl_tick}") if tp != tp_tick: print(f" TP округлен: {tp} → {tp_tick}") except Exception as e: errors.append(f"Ошибка округления цен: {e}") # Проверка размера лота try: lot_size = get_lot_size(ticker) print(f" Размер лота: {lot_size} шт") except Exception as e: errors.append(f"Ошибка получения размера лота: {e}") return len(errors) == 0, errors def create_trade_orders( ticker: str, direction: str, entry: float, sl: float, tp: float, capital: float, risk_pct: float = 1.0, confidence: int = 0, source: str = "manual" ) -> dict: """Создать торговые ордера и логировать сделку.""" print(f"\n🎯 Создание сделки для {ticker}...") print(f" Направление: {direction}") print(f" Вход: {entry:.2f} ₽") print(f" SL: {sl:.2f} ₽") print(f" TP: {tp:.2f} ₽") print(f" Риск: {risk_pct}% от капитала") print(f" Капитал: {format_currency(capital)}") # Валидация параметров valid, errors = validate_trade_parameters(ticker, direction, entry, sl, tp) if not valid: print("❌ Ошибки валидации:") for error in errors: print(f" - {error}") return {"success": False, "errors": errors} # Получаем FIGI try: figi = get_figi(ticker) except ValueError as e: print(f"❌ Ошибка получения FIGI: {e}") return {"success": False, "errors": [str(e)]} # Расчёт размера позиции try: sizing = calculate_position_size(ticker, direction, entry, sl, tp, capital, risk_pct) print(f"\n📊 Расчёт позиции:") print(f" Лотов: {sizing.quantity_lots}") print(f" Акций: {sizing.quantity_shares}") print(f" Риск: {format_currency(sizing.risk_rub)} ({sizing.risk_pct:.2f}%)") print(f" RR: {sizing.reward_risk_ratio:.2f} (без комиссий)") print(f" RR с комиссиями: {sizing.reward_risk_ratio_net:.2f}") # Проверка достаточности капитала if sizing.risk_rub > capital * risk_pct / 100 * 1.2: print(f"⚠️ Риск с комиссиями превышает лимит!") return {"success": False, "errors": ["Недостаточно капитала для сделки"]} except Exception as e: print(f"❌ Ошибка расчёта позиции: {e}") return {"success": False, "errors": [str(e)]} # Создание ордеров orders = {} try: # Входной ордер print(f"\n📝 Создание входного ордера...") entry_order = place_limit_order( ticker=ticker, direction=direction, quantity_lots=sizing.quantity_lots, price_rub=sizing.entry_price_tick ) if "error" in entry_order: print(f"❌ Ошибка создания входного ордера: {entry_order['error']}") return {"success": False, "errors": [entry_order['error']]} orders["entry"] = entry_order print(f" ✅ Входной ордер создан: {entry_order.get('orderId', 'N/A')[:12]}...") # Стоп-лосс ордер print(f"\n🛑 Создание стоп-лосса...") sl_direction = "SELL" if direction == "BUY" else "BUY" sl_order = place_stop_order( ticker=ticker, direction=sl_direction, quantity_lots=sizing.quantity_lots, stop_price_rub=sizing.sl_price_tick, limit_price_rub=sizing.sl_price_tick ) if "error" in sl_order: print(f"❌ Ошибка создания SL ордера: {sl_order['error']}") # Попытка отменить входной ордер if "orderId" in entry_order: from src.api.tbank import cancel_order cancel_order(entry_order["orderId"]) return {"success": False, "errors": [sl_order['error']]} orders["sl"] = sl_order print(f" ✅ SL ордер создан: {sl_order.get('stopOrderId', 'N/A')[:12]}...") # Тейк-профит ордер print(f"\n🎯 Создание тейк-профита...") tp_direction = "SELL" if direction == "BUY" else "BUY" tp_order = place_stop_order( ticker=ticker, direction=tp_direction, quantity_lots=sizing.quantity_lots, stop_price_rub=sizing.tp_price_tick, stop_order_type="STOP_ORDER_TYPE_TAKE_PROFIT" ) if "error" in tp_order: print(f"❌ Ошибка создания TP ордера: {tp_order['error']}") # Попытка отменить предыдущие ордера if "stopOrderId" in sl_order: cancel_order(sl_order["stopOrderId"]) if "orderId" in entry_order: cancel_order(entry_order["orderId"]) return {"success": False, "errors": [tp_order['error']]} orders["tp"] = tp_order print(f" ✅ TP ордер создан: {tp_order.get('stopOrderId', 'N/A')[:12]}...") except Exception as e: print(f"❌ Ошибка создания ордеров: {e}") return {"success": False, "errors": [str(e)]} # Логирование сделки try: print(f"\n📝 Логирование сделки...") trade = journal.open_trade( ticker=ticker, figi=figi, direction=direction, entry_price=sizing.entry_price, entry_price_tick=sizing.entry_price_tick, quantity_lots=sizing.quantity_lots, quantity_shares=sizing.quantity_shares, entry_commission=sizing.commission_entry, sl_price=sizing.sl_price, tp_price=sizing.tp_price, capital=capital, risk_pct=sizing.risk_pct, planned_risk=sizing.risk_rub, planned_reward=sizing.reward_rub, planned_rr=sizing.reward_risk_ratio, confidence=confidence, source=source, entry_order_id=entry_order.get("orderId", "") ) print(f" ✅ Сделка залогирована: {trade.trade_id[:8]}...") # Сохраняем ID ордеров в сделку trade["sl_order_id"] = sl_order.get("stopOrderId", "") trade["tp_order_id"] = tp_order.get("stopOrderId", "") journal._save() # Сохраняем изменения return { "success": True, "trade": trade, "orders": orders, "sizing": sizing } except Exception as e: print(f"❌ Ошибка логирования сделки: {e}") # Попытка отменить все ордера for order_type, order in orders.items(): if "orderId" in order: cancel_order(order["orderId"]) elif "stopOrderId" in order: cancel_order(order["stopOrderId"]) return {"success": False, "errors": [str(e)]} def main(): parser = argparse.ArgumentParser( description="Создание торговых ордеров на основе подготовленных параметров" ) parser.add_argument("--ticker", required=True, help="Тикер инструмента") parser.add_argument("--direction", required=True, choices=["BUY", "SELL"], help="Направление сделки") parser.add_argument("--entry", type=float, required=True, help="Цена входа") parser.add_argument("--sl", type=float, required=True, help="Цена стоп-лосса") parser.add_argument("--tp", type=float, required=True, help="Цена тейк-профита") parser.add_argument("--capital", type=float, required=True, help="Капитал в рублях") parser.add_argument("--risk", type=float, default=1.0, help="Риск в % от капитала (по умолч. 1.0)") parser.add_argument("--confidence", type=int, default=0, help="Уверенность в сигнале (0-100)") parser.add_argument("--env", choices=["sandbox", "real"], default="sandbox", help="Окружение T-Bank API") args = parser.parse_args() # Установка окружения print(f"🔄 Установка окружения: {args.env}...") set_tbank_env(args.env) # Проверка баланса balance_info = check_balance() if not balance_info: print("❌ Не удалось проверить баланс. Прерывание.") sys.exit(1) # Проверка достаточности капитала available_balance = balance_info["available_balance"] required_capital = args.capital if available_balance < required_capital: print(f"⚠️ Доступный баланс {format_currency(available_balance)} меньше требуемого капитала {format_currency(required_capital)}") response = input("Продолжить с доступным балансом? (y/N): ").lower() if response != 'y': print("❌ Прерывание пользователем") sys.exit(1) required_capital = available_balance # Создание сделки result = create_trade_orders( ticker=args.ticker, direction=args.direction, entry=args.entry, sl=args.sl, tp=args.tp, capital=required_capital, risk_pct=args.risk, confidence=args.confidence, source="manual" ) if result["success"]: print(f"\n🎉 Сделка успешно создана!") print(f" Тикер: {result['trade'].ticker}") print(f" Направление: {result['trade'].direction}") print(f" Вход: {result['trade'].entry_price_tick:.2f} ₽") print(f" SL: {result['trade'].sl_price:.2f} ₽") print(f" TP: {result['trade'].tp_price:.2f} ₽") print(f" Лотов: {result['trade'].entry_quantity_lots}") print(f" Риск: {format_currency(result['trade'].planned_risk_rub)} ({result['trade'].risk_pct:.2f}%)") print(f" RR: {result['trade'].planned_rr:.2f}") print(f" ID сделки: {result['trade'].trade_id[:8]}...") if args.env == "real": print(f"\n⚠️ ВНИМАНИЕ: Сделка создана на РЕАЛЬНОМ счёте!") print(f" Это реальная торговля с реальными деньгами.") print(f" Все ордера активны и будут исполнены по рыночным условиям.") else: print(f"\n✅ Сделка создана в тестовом режиме (sandbox)") else: print(f"\n❌ Не удалось создать сделку:") for error in result["errors"]: print(f" - {error}") sys.exit(1) if __name__ == "__main__": main()