Repository navigation
Expand file tree
/
Copy pathmain.py
More file actions
842 lines (708 loc) · 33 KB
/
Copy pathmain.py
File metadata and controls
842 lines (708 loc) · 33 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
3
8000
2
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
"""
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}"
def _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()