miniQMT/main.py at main · laven86/miniQMT

Repository navigation

32

"""

QMT量化交易系统主程序

"""

import os

import time

import threading

import signal

import sys

from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeoutError

from datetime import datetime

# ── 账号专属配置预处理（必须在 import config 之前执行）──

# 支持命令行: python main.py --account-id TEST_ACC_1

_argv = sys.argv[1:]

for _i, _arg in enumerate(_argv):

if _arg in ('--account-id', '--account') and _i + 1 < len(_argv):

os.environ.setdefault('QMT_ACCOUNT_ID', _argv[_i + 1])

break

if _arg.startswith('--account-id='):

os.environ.setdefault('QMT_ACCOUNT_ID', _arg.split('=', 1)[1])

break

# ─────────────────────────────────────────────────────────

import config

# ── 设置控制台窗口标题 + 写入 PID 文件（供 menu.bat / scripts/_launcher.py 跟踪进程）──

def _setup_window_and_pid():

_aid = os.environ.get('QMT_ACCOUNT_ID', '').strip()

if sys.platform == 'win32' and _aid:

try:

import ctypes

ctypes.windll.kernel32.SetConsoleTitleW(f"miniQMT [{_aid}]")

except Exception:

pass

try:

os.makedirs(config.DATA_DIR, exist_ok=True)

pid_file = os.path.join(config.DATA_DIR, "pid.txt")

with open(pid_file, 'w', encoding='ascii') as f:

f.write(str(os.getpid()))

import atexit

def _cleanup_pid():

try:

if os.path.exists(pid_file):

os.remove(pid_file)

except Exception:

pass

atexit.register(_cleanup_pid)

except Exception:

pass

_setup_window_and_pid()

from logger import get_logger, schedule_log_cleanup, clean_old_logs

# 获取logger

logger = get_logger("main")

# 全局变量（threads 须先于文件桥定义：桥启动时登记自己的停止函数）

threads = []

def start_file_bridge_thread():

"""启动大QMT文件桥守护线程（FileRelayDaemon 嵌入本进程）。

沙箱封禁大QMT策略进程一切 TCP 外连后，文件桥是唯一可用 RPC 通道：

大QMT策略端把 Redis 命令写成请求文件，本线程（桥）轮询读取、代为

执行、写回响应文件。详见 qmt-trader/大QMT文件桥通道方案.md。

失败降级策略：Redis 连不上或锁冲突（另一实例在跑且自愈失败）时

仅告警不阻断主流程（桥不启动，交易通道不可用会在业务侧自行报错）；

在业务模块导入之前启动（桥的 prepare() 会清空在途 req/resp，且

PositionManager 导入期就会 ping 大QMT）。

"""

if not getattr(config, "ENABLE_QMT_FILE_BRIDGE", False):

logger.info("文件桥未启用 (ENABLE_QMT_FILE_BRIDGE=False)")

return

# 照抄 position_manager._create_qmt_trader 的 sys.path 先例：插入

# qmt-trader 目录后 import 桥模块。线程化后桥还能继承本进程

# config._load_dotenv_fallback 注入的环境变量（独立启动吃不到 .env）。

import os as _os

import sys as _sys

_bridge_dir = _os.path.join(_os.path.dirname(_os.path.abspath(__file__)), "qmt-trader")

if _bridge_dir not in _sys.path:

_sys.path.insert(0, _bridge_dir)

try:

from qmt_relay_common import build_redis_client

from qmt_file_relay import FileRelayDaemon, FileRelayLockedError

except Exception as e:

logger.warning(f"⚠️ 文件桥模块加载失败:{str(e)[:50]}，主流程继续")

return

# Redis 连接复用 QMT_RPC_REDIS；ping 重试失败告警降级不阻断主流程。

rpc = config.QMT_RPC_REDIS

client = build_redis_client(

rpc.get("host", "127.0.0.1"),

rpc.get("port", 6379),

rpc.get("db", 5),

rpc.get("password", ""),

)

for attempt in range(1, 6):

try:

client.ping()

break

except Exception as exc:

logger.warning(f"文件桥 Redis 未就绪({str(exc)[:40]})，重试 {attempt}/5")

time.sleep(1.5)

else:

logger.warning("⚠️ 文件桥连不上 Redis，桥不启动，主流程继续")

return

daemon = FileRelayDaemon(

config.QMT_FILE_DIR,

client,

token=getattr(config, "QMT_FILE_TOKEN", ""),

key_prefix=getattr(config, "QMT_FILE_KEY_PREFIX", "bigqmt:"),

poll_interval=getattr(config, "QMT_FILE_POLL_INTERVAL", 0.002),

)

def _serve():

try:

daemon.serve_forever()

except FileRelayLockedError as exc:

# 桥内已自愈一次（锁内 pid 已死则清锁重试）；仍失败说明确有

# 活实例在跑（或锁不可删），降级：桥不启动，主流程继续。

logger.warning(f"⚠️ 文件桥锁冲突，桥不启动:{str(exc)[:80]}")

except Exception as exc:

logger.error(f"文件桥线程异常退出:{str(exc)[:50]}")

thread = threading.Thread(target=_serve, name="FileBridge", daemon=True)

thread.start()

# 等待就绪或线程退出（锁冲突会立即抛出结束线程）：prepare() 清空在途

# req/resp，ready 置位前业务侧不应发请求。

deadline = time.time() + 5

while time.time() < deadline:

if daemon.ready.is_set() or not thread.is_alive():

break

time.sleep(0.05)

if daemon.ready.is_set():

def _stop_file_bridge():

# 先 stop 再 join（timeout=2）：清理路径中由 threads 遍历统一调用。

daemon.stop()

if thread.is_alive():

thread.join(timeout=2)

threads.append(("file_bridge", _stop_file_bridge))

logger.info(f"✅ 文件桥已启动 (dir:{config.QMT_FILE_DIR})")

else:

if thread.is_alive():

daemon.stop()

thread.join(timeout=2)

logger.warning("⚠️ 文件桥启动失败（未就绪），降级为不启动，主流程继续")

# 文件桥必须早于下面的业务模块导入：web_server 导入时经 get_trading_executor()

# 创建 PositionManager 并同步 ping 大QMT，桥不在线则 ping 必然超时、启动转离线。

# 仅直接运行时启动，测试 import main 不拉起桥线程。

if __name__ == "__main__":

start_file_bridge_thread()

from maintenance import schedule_database_maintenance

from data_manager import get_data_manager

from indicator_calculator import get_indicator_calculator

from position_manager import get_position_manager

from trading_executor import get_trading_executor

from strategy import get_trading_strategy

from web_server import start_web_server

from config_manager import get_config_manager

from thread_monitor import get_thread_monitor

# 全局变量

stop_event = threading.Event()

system_start_time = None # 系统启动时间

heartbeat_thread = None # 心跳日志线程

# ── spinner（旋转指示器）──

_spinner_stop = threading.Event()

_spinner_thread = None

def _switch_text(enabled):

return "开启" if enabled else "关闭"

def _get_existing_position_manager():

position_manager_module = sys.modules.get("position_manager")

return getattr(position_manager_module, "_instance", None)

def _get_runtime_identity(position_manager=None):

"""返回当前进程身份信息，清理阶段不得因取值失败影响退出。"""

try:

account_config = config.get_account_config()

account_id = os.environ.get("QMT_ACCOUNT_ID", "").strip() or account_config.get("account_id", "")

except Exception:

account_id = os.environ.get("QMT_ACCOUNT_ID", "").strip()

session_id = "N/A"

try:

pm = position_manager or _get_existing_position_manager()

qmt_trader = getattr(pm, "qmt_trader", None) if pm else None

qmt_config = getattr(qmt_trader, "config", None)

session_id = (

getattr(qmt_trader, "session_id", None)

or getattr(qmt_config, "session_id", None)

or "N/A"

)

except Exception:

session_id = "N/A"

return {

"pid": os.getpid(),

"account_id": account_id or "未配置",

"port": getattr(config, "WEB_SERVER_PORT", "N/A"),

"session_id": session_id,

}

def _format_runtime_identity(identity=None):

identity = identity or _get_runtime_identity()

return (

f"pid={identity['pid']} "

f"account_id={identity['account_id']} "

f"port={identity['port']} "

f"session_id={identity['session_id']}"

)

def _log_process_lifecycle(event, position_manager=None):

logger.info(f"进程生命周期: {event} {_format_runtime_identity(_get_runtime_identity(position_manager))}")

def _get_active_grid_session_count(position_manager=None):

try:

pm = position_manager or _get_existing_position_manager()

grid_manager = getattr(pm, "grid_manager", None) if pm else None

if not grid_manager:

return 0

lock = getattr(grid_manager, "lock", None)

if lock:

with lock:

sessions = list((getattr(grid_manager, "sessions", {}) or {}).values())

else:

sessions = list((getattr(grid_manager, "sessions", {}) or {}).values())

return sum(

1 for session in sessions

if getattr(session, "status", None) == "active" and getattr(session, "enabled", True)

)

except Exception:

return "获取失败"

def _format_heartbeat_status_lines(active_grid_session_count):

return (

f" 模式:{'模拟' if config.ENABLE_SIMULATION_MODE else '实盘'} | "

f"无人值守总开关:{_switch_text(config.ENABLE_AUTO_OPERATION)} | "

f"自动止盈:{_switch_text(config.ENABLE_AUTO_TRADING)}",

f" 自动网格:{_switch_text(config.ENABLE_GRID_TRADING)} | "

f"活跃网格会话数:{active_grid_session_count}",

)

def _format_resource_line():

"""进程资源占用行：线程数（Python + OS）、句柄数与内存。

长跑进程线程/内存持续增长会以 RuntimeError: can't start new thread 收场，

且进程静默死亡后无任何线索，因此把这些指标写进心跳便于事后归因。

线程数同时给出两个口径：threading.active_count() 只覆盖 Python 层，

xtquant / QMT SDK 的原生线程只体现在 OS 口径里（实测 18 vs 104）。

句柄数用于识别句柄泄漏——耗尽时同样表现为打开文件失败。

"""

thread_count = threading.active_count()

try:

from utils import memory_usage, process_resource_stats

mem = memory_usage()

proc = process_resource_stats()

except Exception:

mem = None

proc = None

if proc:

thread_str = f"线程数:{thread_count}(OS {proc['os_threads']}) | 句柄:{proc['handles']}"

else:

thread_str = f"线程数:{thread_count}"

if mem:

mem_str = f"内存:RSS {mem['rss_mb']:.0f}MB / VMS {mem['vms_mb']:.0f}MB"

else:

mem_str = "内存:获取失败"

return f" {thread_str} | {mem_str}"

d 8000 ef _spinner_worker():

"""在终端同一行滚动显示 |/-\\ 旋转符号，表示程序正在运行。

直接写 stdout，不进入日志文件；logger 输出的完整行会自然覆盖该符号。

刷新间隔取 config.SPINNER_INTERVAL：这是本进程写 stdout 最频繁的来源，

而 Windows Terminal 在高频写入下会无限泄漏内存，频率直接决定泄漏速度。

"""

chars = r'|/-\\'

interval = getattr(config, 'SPINNER_INTERVAL', 1.0)

i = 0

while not _spinner_stop.is_set():

sys.stdout.write('\r' + chars[i % 4] + ' ')

sys.stdout.flush()

i += 1

_spinner_stop.wait(interval)

sys.stdout.write('\r \r') # 退出时清除残留字符

sys.stdout.flush()

def signal_handler(sig, frame):

"""信号处理函数，用于捕获退出信号"""

logger.info("收到退出信号")

stop_event.set()

sys.exit(0)

def _check_stop_signal():

"""检查 launcher 写入的停止信号文件，实现跨控制台的优雅关闭。

Windows 下 CREATE_NEW_CONSOLE 创建的进程有独立控制台，

GenerateConsoleCtrlEvent 无法送达 Ctrl+C。

launcher 的 cmd_stop 在发送 Ctrl+C 的同时，也会写入

data_<id>/stop_signal 文件。主循环检测到该文件即触发优雅退出。

"""

signal_path = os.path.join(config.DATA_DIR, "stop_signal")

if os.path.exists(signal_path):

logger.info("检测到停止信号文件，触发优雅退出")

try:

os.remove(signal_path)

except Exception:

pass

stop_event.set()

def load_persisted_configs():

"""从数据库加载持久化配置"""

logger.info("加载持久化配置")

try:

config_manager = get_config_manager()

applied_count = config_manager.apply_configs_to_runtime()

logger.info(f"✓ 配置{applied_count}项")

return applied_count

except Exception as e:

logger.error(f"配置加载失败:{str(e)[:30]}")

return 0

def _start_xtquant_manager_server():

"""

启动 XtQuantManager HTTP 服务（后台线程）。

仅在 config.ENABLE_XTQUANT_MANAGER=True 时生效。

服务启动后，position_manager 和 data_manager 的工厂函数

可通过 HTTP 路由到该服务，实现多账号统一管理。

Returns:

XtQuantServer: 已启动的服务实例；ENABLE_XTQUANT_MANAGER=False 时返回 None。

"""

if not getattr(config, "ENABLE_XTQUANT_MANAGER", False):

return None

try:

from xtquant_manager import XtQuantServer, XtQuantServerConfig, XtQuantManager

from xtquant_manager.account import AccountConfig

xqm_port = getattr(config, "XTQUANT_MANAGER_PORT", 8888)

srv_cfg = XtQuantServerConfig(

host="127.0.0.1",

port=xqm_port,

api_token=getattr(config, "XTQUANT_MANAGER_TOKEN", ""),

rate_limit=getattr(config, "XTQUANT_MANAGER_RATE_LIMIT", 600),

)

server = XtQuantServer(config=srv_cfg)

server.start(blocking=False)

logger.info(f"✓ XtQuantManager HTTP 服务已启动 (127.0.0.1:{xqm_port})")

# 注册所有账号（支持单账号和多账号两种配置格式）

manager = XtQuantManager.get_instance()

accounts_list = config.get_all_accounts_config()

for acfg in accounts_list:

acc_id = acfg.get("account_id", "")

if not acc_id:

continue

try:

ok = manager.register_account(AccountConfig(

account_id=acc_id,

qmt_path=acfg.get("qmt_path", config.QMT_PATH),

account_type=acfg.get("account_type", "STOCK"),

))

if ok:

logger.info(f"✓ 账号 {acc_id[:4]}*** 注册成功")

else:

logger.warning(f"账号 {acc_id[:4]}*** 注册但连接失败，将自动重连")

except Exception as e:

logger.error(f"注册账号 {acc_id[:4]}*** 失败: {e}")

logger.info(f"✓ 共注册 {len(accounts_list)} 个账号")

return server

except Exception as e:

logger.error(f"XtQuantManager 服务启动失败: {e}")

return None

def init_system():

"""初始化系统"""

logger.info("系统初始化")

# 创建数据目录

if not os.path.exists(config.DATA_DIR):

os.makedirs(config.DATA_DIR)

logger.info(f"✓ 创建目录:{config.DATA_DIR}")

# 加载持久化配置（在初始化其他模块之前）

load_persisted_configs()

# 按需启动 XtQuantManager HTTP 服务（在各模块初始化之前）

_start_xtquant_manager_server()

# 获取各个模块的实例

data_manager = get_data_manager()

indicator_calculator = get_indicator_calculator()

position_manager = get_position_manager()

trading_executor = get_trading_executor()

trading_strategy = get_trading_strategy()

# Fix: web_server.py 模块加载时（import 阶段）触发 DataManager 单例创建，

# 但此时 XtQuantManager HTTP 服务尚未启动，导致 xtdata 初始化失败（self.xt=None）。

# 在 HTTP 服务启动后，调用 reinit_xtquant() 补充初始化。

if getattr(config, "ENABLE_XTQUANT_MANAGER", False):

data_manager.reinit_xtquant()

# 预热股票名称缓存，避免首次交易时触发 baostock 导致 xtdata 断连

data_manager.warm_stock_name_cache()

logger.info("✓ 系统初始化完成")

return data_manager, indicator_calculator, position_manager, trading_executor, trading_strategy

def heartbeat_logger():

"""系统心跳日志 - 定期输出运行状态摘要"""

while not stop_event.is_set():

try:

# 等待指定间隔

if stop_event.wait(config.HEARTBEAT_INTERVAL):

break # 收到停止信号

# 计算运行时长

if system_start_time:

uptime = datetime.now() - system_start_time

hours, remainder = divmod(int(uptime.total_seconds()), 3600)

minutes, seconds = divmod(remainder, 60)

uptime_str = f"{hours}h{minutes}m{seconds}s"

else:

uptime_str = "未知"

position_manager = None

# 获取持仓信息

try:

position_manager = get_position_manager()

positions = position_manager.get_all_positions()

position_count = 0 if positions is None or positions.empty else len(positions)

# 获取账户信息

account_info = position_manager.get_account_info()

if account_info:

total_asset = account_info.get('total_asset', 0)

available = account_info.get('available', 0)

asset_str = f"总资产:{total_asset:.2f} 可用:{available:.2f}"

else:

asset_str = "资产信息获取失败"

except Exception as e:

position_count = "获取失败"

asset_str = f"获取失败:{str(e)[:20]}"

# 输出心跳日志

try:

market_health_str = get_data_manager().format_market_health_summary()

except Exception as e:

market_health_str = f"行情健康: 获取失败:{str(e)[:20]}"

active_grid_session_count = _get_active_grid_session_count(position_manager)

logger.info("=" * 50)

logger.info(f"💓 系统心跳 - 运行时长:{uptime_str}")

for status_line in _format_heartbeat_status_lines(active_grid_session_count):

logger.info(status_line)

logger.info(f" 持仓数量:{position_count} | {asset_str}")

logger.info(_format_resource_line())

logger.info(f" {market_health_str}")

logger.info("=" * 50)

except Exception as e:

logger.error(f"心跳日志出错:{str(e)[:50]}")

time.sleep(60) # 出错后等待一分钟再继续

def start_heartbeat_logger():

"""启动心跳日志线程及终端旋转指示器"""

global heartbeat_thread, system_start_time, _spinner_thread

if not config.ENABLE_HEARTBEAT_LOG:

logger.info("心跳日志未启用 (ENABLE_HEARTBEAT_LOG=False)")

return

system_start_time = datetime.now()

heartbeat_thread = threading.Thread(target=heartbeat_logger, daemon=True, name="HeartbeatLogger")

heartbeat_thread.start()

_spinner_stop.clear()

_spinner_thread = threading.Thread(target=_spinner_worker, daemon=True, name="SpinnerThread")

_spinner_thread.start()

logger.info(f"✅ 心跳日志已启动 (间隔:{config.HEARTBEAT_INTERVAL}秒)")

def stop_heartbeat_logger():

"""停止心跳日志线程及旋转指示器"""

global heartbeat_thread, _spinner_thread

_spinner_stop.set()

if _spinner_thread and _spinner_thread.is_alive():

_spinner_thread.join(timeout=1)

if heartbeat_thread and heartbeat_thread.is_alive():

logger.info("停止心跳日志线程")

stop_event.set()

heartbeat_thread.join(timeout=2)

def start_data_thread(data_manager):

"""启动数据更新线程"""

if config.ENABLE_DATA_SYNC:

logger.info("启动数据更新线程")

data_manager.start_data_update_thread()

threads.append(("data_thread", data_manager.stop_data_update_thread))

def start_position_thread(position_manager):

"""启动持仓监控线程"""

if config.ENABLE_POSITION_MONITOR:

logger.info("启动持仓监控")

position_manager.start_position_monitor_thread()

# 🔑 验证线程启动

time.sleep(0.5) # 等待线程启动

if position_manager.monitor_thread and position_manager.monitor_thread.is_alive():

logger.info("✅ 持仓监控已启动")

else:

logger.error("❌ 持仓监控启动失败")

threads.append(("position_thread", position_manager.stop_position_monitor_thread))

else:

logger.warning("⚠️ 持仓监控未启用")

def start_strategy_thread(trading_strategy):

"""启动策略线程"""

# if config.ENABLE_AUTO_TRADING:

logger.info("启动策略线程")

trading_strategy.start_strategy_thread()

threads.append(("strategy_thread", trading_strategy.stop_strategy_thread))

def start_log_cleanup_thread():

"""启动日志清理线程"""

if config.ENABLE_LOG_CLEANUP:

logger.info("启动日志清理线程")

log_thread = threading.Thread(target=schedule_log_cleanup)

log_thread.daemon = True

log_thread.start()

threads.append(("log_thread", lambda: None)) # 没有停止函数，依赖于daemon=True

def start_database_maintenance_thread():

"""启动数据库维护线程"""

if config.ENABLE_DB_MAINTENANCE:

logger.info("启动数据库维护线程")

db_maintenance_thread = threading.Thread(target=schedule_database_maintenance)

db_maintenance_thread.daemon = True

db_maintenance_thread.start()

threads.append(("db_maintenance_thread", lambda: None))

def start_settlement_snapshot_thread(position_manager):

"""启动收盘快照线程（持仓 + 净值）。

与 db_maintenance 同为"每日定时轮询"范式：过了点就补、当天只跑一次，

因此进程收盘后才启动也能补录当天快照。

"""

if not getattr(config, 'ENABLE_SETTLEMENT_SNAPSHOT', True):

logger.info("收盘快照线程未启用")

return

logger.info("启动收盘快照线程")

import settlement_db

settlement_db.ensure_schema()

snapshot_thread = threading.Thread(

target=lambda: settlement_db.schedule_close_snapshot(position_manager, stop_event))

snapshot_thread.daemon = True

snapshot_thread.start()

threads.append(("settlement_snapshot_thread", lambda: None))

def start_web_server_thread(position_manager):

"""启动Web服务器线程

Args:

position_manager: 已初始化的position_manager实例

"""

logger.info("启动Web服务器线程")

# logger.info(f"[DEBUG main.py] 传入Web服务器的position_manager id: {id(position_manager)}")

# 创建线程并传入position_manager

web_thread = threading.Thread(target=lambda: start_web_server(position_manager))

web_thread.daemon = True

web_thread.start()

# 使用shutdown_web_server进行资源清理

from web_server import shutdown_web_server

threads.append(("web_thread", shutdown_web_server))

def download_initial_data(data_manager):

"""下载初始数据，每只股票有超时保护"""

logger.info("下载初始数据")

timeout = config.HISTORY_DATA_DOWNLOAD_TIMEOUT

for stock_code in config.STOCK_POOL:

try:

logger.info(f"下载 {stock_code[:6]} 历史数据（超时{timeout}秒）")

executor = ThreadPoolExecutor(max_workers=1)

future = executor.submit(data_manager.download_history_data, stock_code)

try:

data_df = future.result(timeout=timeout)

if data_df is not None and not data_df.empty:

data_manager.save_history_data(stock_code, data_df)

except FuturesTimeoutError:

logger.warning(f"下载 {stock_code[:6]} 超时（>{timeout}秒），跳过")

finally:

executor.shutdown(wait=False) # 不阻塞等待后台线程，立即继续下一只

except Exception as e:

logger.error(f"下载 {stock_code[:6]} 失败:{str(e)[:30]}")

# 避免请求过于频繁

time.sleep(1)

logger.info("初始数据下载完成")

def calculate_initial_indicators(indicator_calculator):

"""计算初始指标"""

logger.info("计算初始指标")

indicator_calculator.update_all_stock_indicators()

logger.info("初始指标计算完成")

def cleanup():

"""清理资源 - 优雅关闭版本"""

_log_process_lifecycle("开始清理")

# 第1步: 先停止Web服务器(避免在关闭数据库后仍有请求)

for thread_name, stop_func in threads:

if thread_name == "web_thread":

try:

logger.info("停止Web服务器")

stop_func()

except Exception as e:

logger.error(f"Web服务器停止失败:{str(e)[:30]}")

break

# 第2步: 停止线程监控器(如果启用)

if config.ENABLE_THREAD_MONITOR:

try:

logger.info("停止线程监控")

thread_monitor = get_thread_monitor()

thread_monitor.stop()

except Exception as e:

logger.error(f"线程监控停止失败:{str(e)[:30]}")

# 第2.5步: 停止心跳日志线程

try:

stop_heartbeat_logger()

except Exception as e:

logger.error(f"心跳日志停止失败:{str(e)[:30]}")

# 第3步: 停止其他业务线程

for thread_name, stop_func in threads:

if thread_name == "web_thread":

continue # 已经停止

try:

logger.info(f"停止{thread_name}")

stop_func()

except Exception as e:

logger.error(f"{thread_name}停止失败:{str(e)[:30]}")

# 第4步: 关闭各个模块(按依赖顺序)

try:

trading_strategy = get_trading_strategy()

trading_strategy.close()

except Exception as e:

logger.error(f"策略关闭失败:{str(e)[:30]}")

try:

trading_executor = get_trading_executor()

trading_executor.close()

except Exception as e:

logger.error(f"执行器关闭失败:{str(e)[:30]}")

try:

data_manager = get_data_manager()

data_manager.close()

except Exception as e:

logger.error(f"数据管理器关闭失败:{str(e)[:30]}")

_log_process_lifecycle("清理完成")

def main():

"""主函数"""

try:

logger.info("=" * 50)

logger.info(f"QMT系统启动 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")

logger.info("=" * 50)

# 设置信号处理

signal.signal(signal.SIGINT, signal_handler)

signal.signal(signal.SIGTERM, signal_handler)

# 初始化系统

data_manager, indicator_calculator, position_manager, trading_executor, trading_strategy = init_system()

_log_process_lifecycle("启动", position_manager)

# 初始化网格交易管理器

logger.info(f"检查网格交易配置: ENABLE_GRID_TRADING = {config.ENABLE_GRID_TRADING}")

# logger.info(f"[DEBUG main.py] position_manager id: {id(position_manager)}")

# logger.info(f"[DEBUG main.py] position_manager有grid_manager属性: {hasattr(position_manager, 'grid_manager')}")

if config.ENABLE_GRID_TRADING:

try:

logger.info("初始化网格交易管理器")

# logger.info(f"[DEBUG main.py] 调用init_grid_manager前，grid_manager={getattr(position_manager, 'grid_manager', 'NO_ATTR')}")

position_manager.init_grid_manager(trading_executor)

# logger.info(f"[DEBUG main.py] 调用init_grid_manager后，grid_manager={position_manager.grid_manager}")

# logger.info(f"[DEBUG main.py] grid_manager类型: {type(position_manager.grid_manager)}")

logger.info("✓ 网格交易管理器初始化完成")

# ⚠️ 新增: 初始化风险等级模板(仅首次运行或更新时)

try:

logger.info("开始初始化风险等级模板...")

# 从position_manager获取db_manager

db_mgr = position_manager.db_manager

initialized_count = db_mgr.init_risk_level_templates()

if initialized_count > 0:

logger.info(f"✓ 新增 {initialized_count} 个风险等级模板")

else:

logger.info("✓ 风险等级模板已存在,无需初始化")

except Exception as e:

logger.warning(f"初始化风险等级模板失败(不影响系统运行): {str(e)}")

except Exception as e:

logger.error(f"网格交易管理器初始化失败: {str(e)}")

logger.info("系统继续运行(网格交易功能不可用)")

else:

logger.warning("网格交易功能未启用 (ENABLE_GRID_TRADING=False)")

# 下载初始数据

download_initial_data(data_manager)

# 计算初始指标

calculate_initial_indicators(indicator_calculator)

# 启动各个线程（文件桥已在本模块顶层启动）

start_data_thread(data_manager)

start_position_thread(position_manager)

start_strategy_thread(trading_strategy)

start_log_cleanup_thread()

start_database_maintenance_thread()

start_settlement_snapshot_thread(position_manager)

# ============ 新增: 启动盘前同步调度器 ============

from premarket_sync import start_premarket_sync_scheduler

start_premarket_sync_scheduler()

logger.info("✓ 盘前同步调度器已启动")

# ============ 新增: 启动线程健康监控 ============

if config.ENABLE_THREAD_MONITOR:

thread_monitor = get_thread_monitor()

# 注册持仓监控线程

# 注意：不注册 heartbeat_check=check_qmt_connection_health。

# QMT 断连 ≠ 线程崩溃——持仓监控线程本身始终健康，

# QMT 重连逻辑由 _attempt_qmt_reconnect() 在监控循环内部自主处理。

# 若注册 heartbeat_check，ping 失败会触发 _restart_thread()，

# 而线程仍存活导致 start 幂等返回，产生误导性"重启成功"日志噪音。

thread_monitor.register_thread(

"持仓监控线程",

lambda: position_manager.monitor_thread,

position_manager.start_position_monitor_thread,

)

# 注册数据更新线程

thread_monitor.register_thread(

"数据更新线程",

lambda: data_manager.update_thread,

data_manager.start_data_update_thread

)

# 注册策略线程

thread_monitor.register_thread(

"策略线程",

lambda: trading_strategy.strategy_thread,

trading_strategy.start_strategy_thread

)

# 启动监控

thread_monitor.start()

logger.info("✅ 线程监控已启动")

# ============ 新增: 启动系统心跳日志 ============

start_heartbeat_logger()

# ============ 新增: 启动卖出监控器 ============

if hasattr(config, 'ENABLE_SELL_MONITOR') and config.ENABLE_SELL_MONITOR:

try:

from sell_monitor import get_sell_monitor

sell_monitor = get_sell_monitor()

logger.info("✅ 卖出监控器已启动")

logger.info(f" 监控:{'启用' if sell_monitor.monitoring_enabled else '禁用'}")

logger.info(f" 告警:{'启用' if config.ENABLE_SELL_ALERT_NOTIFICATION else '禁用'}")

except Exception as e:

logger.warning(f"⚠️ 卖出监控器失败:{str(e)[:30]}")

logger.info("系统继续运行")

# 启动Web服务器。

# web2.0 模式下界面由 xtquant_manager 托管，但 Flask 仍会启动：

# 网关需要反向调用它读取 ENABLE_AUTO_OPERATION 等只存在于本进程

# 内存、不持久化的开关。launcher 仅在端口已被占用时才设 QMT_NO_FLASK。

if not os.environ.get("QMT_NO_FLASK"):

start_web_server_thread(position_manager)

else:

logger.info("跳过 Flask Web 服务器（QMT_NO_FLASK=1，端口已被占用或显式禁用）")

# 等待退出信号

logger.info("✅ 系统启动完成")

while not stop_event.is_set():

_check_stop_signal()

time.sleep(1)

except Exception as e:

logger.error(f"系统运行出错:{str(e)[:30]}")

finally:

cleanup()

logger.info("系统已退出")

if __name__ == "__main__":

main()

FilesExpand file tree

main.py

Latest commit

History

main.py

File metadata and controls