guaiwoluo2020
a705593955
- 新增新闻爬取和监控模块 (news_crawler, news_monitor) - 新增 LLM 分析模块 (llm_analyzer) - 新增持仓管理和交易历史存储 - 新增系统日志功能 - 新增前端页面: News, Positions, Settings, SystemLog - 更新路由和 API 接口 - 更新 .gitignore 排除敏感文件
258 lines
10 KiB
Python
258 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
交易服务核心类
|
|
"""
|
|
|
|
from collections import deque, defaultdict
|
|
from typing import List, Dict, Optional
|
|
import threading
|
|
|
|
from models import TradeInstruction
|
|
from market.store import MarketStore
|
|
from market.pivot_detector import PivotDetector
|
|
from market.monitor import PivotMonitor, TradeConfig
|
|
from market.trend_analyzer import TrendAnalyzer
|
|
from market.pending_orders import PendingOrderManager
|
|
from market.llm_analyzer import LLMAnalyzer
|
|
|
|
|
|
class TradingServer:
|
|
"""交易服务主类"""
|
|
|
|
def __init__(self):
|
|
# 交易指令队列 - 按SYMBOL分类
|
|
# 结构: {"SYMBOL1": [TradeInstruction1, ...], "SYMBOL2": [...]}
|
|
self.trade_instructions = defaultdict(list)
|
|
|
|
# 平仓指令队列 - 按SYMBOL分类
|
|
# 结构: {"SYMBOL1": [ticket1, ticket2, ...], ...}
|
|
self.close_position_instructions = defaultdict(list)
|
|
|
|
# 统计数据历史 - 保留最新10条
|
|
# 结构: deque([{stat_data1}, {stat_data2}, ...], maxlen=10)
|
|
self.statistics_history = deque(maxlen=10)
|
|
|
|
# 线程锁 - 确保线程安全
|
|
self.lock = threading.RLock()
|
|
|
|
# ==================== 行情模块 ====================
|
|
# K线存储
|
|
self.market_store = MarketStore()
|
|
# 转折点检测器
|
|
self.pivot_detector = PivotDetector()
|
|
# 待确认订单管理器（需要在 PivotMonitor 之前初始化）
|
|
self.pending_orders = PendingOrderManager()
|
|
# 设置订单确认回调
|
|
self.pending_orders.set_confirm_callback(self._on_order_confirmed)
|
|
# 大模型分析器（需要在 PivotMonitor 之前初始化）
|
|
self.llm_analyzer = LLMAnalyzer(self.market_store)
|
|
# 转折点监控器
|
|
self.pivot_monitor = PivotMonitor(self.market_store, self.pivot_detector, self.pending_orders, self.llm_analyzer)
|
|
# 设置统计数据历史引用（用于获取价差）
|
|
self.pivot_monitor.set_statistics_history(self.statistics_history)
|
|
# 趋势分析器
|
|
self.trend_analyzer = TrendAnalyzer()
|
|
self.trend_analyzer.set_statistics_history(self.statistics_history)
|
|
# 交易配置
|
|
self.trade_config = TradeConfig.get_instance()
|
|
|
|
print("[信息] 交易服务已初始化")
|
|
|
|
def _on_order_confirmed(self, order: Dict):
|
|
"""
|
|
订单确认回调 - 将确认的订单加入交易队列
|
|
"""
|
|
print(f"[TradingServer] _on_order_confirmed 被调用，订单: {order}")
|
|
try:
|
|
# 创建交易指令
|
|
instruction = TradeInstruction(
|
|
symbol=order.get('symbol', ''),
|
|
action=order.get('action', 'b'),
|
|
mount=order.get('mount', 0.01),
|
|
price=order.get('price', 0),
|
|
sl=order.get('sl', 0),
|
|
tp=order.get('tp', 0),
|
|
description=order.get('description', '')
|
|
)
|
|
print(f"[TradingServer] 创建交易指令: symbol={instruction.symbol}, action={instruction.action}, mount={instruction.mount}, price={instruction.price}, sl={instruction.sl}, tp={instruction.tp}, description={instruction.description}")
|
|
# 添加到交易队列
|
|
result = self.add_trade_instruction([instruction])
|
|
print(f"[TradingServer] 订单已加入交易队列: {result}")
|
|
except Exception as e:
|
|
print(f"[TradingServer] 加入交易队列失败: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
|
|
def add_trade_instruction(self, instructions: List[TradeInstruction]) -> dict:
|
|
"""
|
|
添加交易指令
|
|
返回一个字典，包含添加和拒绝的数量。
|
|
|
|
在此处对缺失的 sl/tp 值进行补全：
|
|
- sl 若未设置保持0.0
|
|
- tp 若未设置则默认 0.005
|
|
|
|
同时按照规则进行价格检查：
|
|
* 买入指令: price 需大于 sl 且小于 tp
|
|
* 卖出指令: price 需小于 sl 且大于 tp
|
|
若 sl 或 tp 未设置（<=0），则忽略检查。
|
|
"""
|
|
with self.lock:
|
|
added = 0
|
|
rejected = 0
|
|
for instruction in instructions:
|
|
# 填充默认值
|
|
if instruction.sl is None:
|
|
instruction.sl = 0.0
|
|
if instruction.tp is None or instruction.tp <= 0.0:
|
|
instruction.tp = 0.005
|
|
|
|
# 验证价格与 SL/TP 关系
|
|
if instruction.sl > 0 and instruction.tp > 0:
|
|
if instruction.action.lower() == 'b':
|
|
if not (instruction.price > instruction.sl and instruction.price < instruction.tp):
|
|
print(f"[警告] 忽略无效买入指令: {instruction}")
|
|
rejected += 1
|
|
continue
|
|
elif instruction.action.lower() == 's':
|
|
if not (instruction.price < instruction.sl and instruction.price > instruction.tp):
|
|
print(f"[警告] 忽略无效卖出指令: {instruction}")
|
|
rejected += 1
|
|
continue
|
|
|
|
# 直接使用原始symbol，不做转换
|
|
symbol = instruction.symbol
|
|
self.trade_instructions[symbol].append(instruction)
|
|
added += 1
|
|
|
|
print(f"[信息] 已添加 {added} 条交易指令, 拒绝 {rejected} 条")
|
|
for symbol, trades in self.trade_instructions.items():
|
|
print(f" {symbol}: {len(trades)} 条待执行")
|
|
|
|
return {"added": added, "rejected": rejected}
|
|
|
|
|
|
def get_trades_by_symbol(self, symbol: str, price: Optional[float] = None) -> Dict:
|
|
"""
|
|
获取指定SYMBOL的交易指令并删除
|
|
|
|
同时调用策略检查（在 PivotMonitor 中执行）
|
|
|
|
返回: {"trades": [...], "pivot_alerts": [...]}
|
|
"""
|
|
# 检查所有策略（关键点位、支撑压力、AI趋势）
|
|
pivot_alerts = []
|
|
if price is not None:
|
|
pivot_alerts = self.pivot_monitor.check_and_alert(symbol, price)
|
|
if pivot_alerts:
|
|
print(f"[信息] {symbol} 当前价格 {price} 接近转折点")
|
|
|
|
with self.lock:
|
|
# 调试：打印当前所有待执行指令
|
|
if len(self.trade_instructions) > 0:
|
|
print(f"[调试] get_trades_by_symbol 查询symbol={symbol}")
|
|
print(f"[调试] 当前trade_instructions keys: {list(self.trade_instructions.keys())}")
|
|
for k, v in self.trade_instructions.items():
|
|
print(f"[调试] {k}: {len(v)} 条")
|
|
|
|
if symbol not in self.trade_instructions or len(self.trade_instructions[symbol]) == 0:
|
|
return {"trades": [], "pivot_alerts": pivot_alerts}
|
|
|
|
# 获取所有指令并直接返回
|
|
trades = self.trade_instructions[symbol]
|
|
result = [{
|
|
"symbol": t.symbol,
|
|
"action": t.action.lower(),
|
|
"mount": t.mount,
|
|
"price": t.price,
|
|
"sl": t.sl,
|
|
"tp": t.tp,
|
|
"description": t.description or ""
|
|
} for t in trades]
|
|
|
|
# 清空指令队列
|
|
self.trade_instructions[symbol] = []
|
|
|
|
if len(result) > 0:
|
|
print(f"[信息] 推送了 {len(result)} 条 {symbol} 指令给EA")
|
|
|
|
return {"trades": result, "pivot_alerts": pivot_alerts}
|
|
|
|
def save_statistics(self, stat_data: dict) -> None:
|
|
"""
|
|
保存统计数据
|
|
自动保留最新10条
|
|
"""
|
|
with self.lock:
|
|
self.statistics_history.append(stat_data)
|
|
print(f"[信息] 统计数据已记录 - {stat_data.get('timestamp', 'unknown')}")
|
|
print(f" 当前保存数据条数: {len(self.statistics_history)}")
|
|
|
|
def get_latest_statistics(self, count: int = 10) -> List[Dict]:
|
|
"""
|
|
获取最新的统计数据
|
|
"""
|
|
with self.lock:
|
|
return list(self.statistics_history)[-count:]
|
|
|
|
def get_all_pending_trades(self) -> Dict[str, List[Dict]]:
|
|
"""
|
|
获取所有待执行的交易指令（不删除）
|
|
用于查询接口
|
|
"""
|
|
with self.lock:
|
|
result = {}
|
|
for symbol, trades in self.trade_instructions.items():
|
|
result[symbol] = [
|
|
{
|
|
"symbol": t.symbol,
|
|
"action": t.action.lower(),
|
|
"mount": t.mount,
|
|
"price": t.price,
|
|
"sl": t.sl,
|
|
"tp": t.tp,
|
|
"description": t.description or ""
|
|
}
|
|
for t in trades
|
|
]
|
|
return result
|
|
|
|
def clear_trades(self, symbol: Optional[str] = None) -> int:
|
|
"""
|
|
清空交易指令
|
|
如果指定symbol则只清空该symbol
|
|
返回清空的指令数量
|
|
"""
|
|
with self.lock:
|
|
if symbol is None:
|
|
total = sum(len(trades) for trades in self.trade_instructions.values())
|
|
self.trade_instructions.clear()
|
|
print(f"[信息] 已清空所有交易指令，共 {total} 条")
|
|
return total
|
|
else:
|
|
count = len(self.trade_instructions.get(symbol, []))
|
|
if symbol in self.trade_instructions:
|
|
del self.trade_instructions[symbol]
|
|
print(f"[信息] 已清空 {symbol} 的交易指令，共 {count} 条")
|
|
return count
|
|
|
|
def add_close_position_instruction(self, symbol: str, ticket: int) -> None:
|
|
"""
|
|
添加平仓指令
|
|
"""
|
|
with self.lock:
|
|
self.close_position_instructions[symbol].append(ticket)
|
|
print(f"[信息] 添加平仓指令: {symbol} ticket={ticket}")
|
|
|
|
def get_close_position_instructions(self, symbol: str) -> List[int]:
|
|
"""
|
|
获取并清空平仓指令
|
|
"""
|
|
with self.lock:
|
|
tickets = self.close_position_instructions.get(symbol, [])
|
|
self.close_position_instructions[symbol] = []
|
|
if tickets:
|
|
print(f"[信息] 返回平仓指令: {symbol} tickets={tickets}")
|
|
return tickets
|