Repository navigation
Expand file tree
/
Copy pathstrategy.py
More file actions
1096 lines (907 loc) · 53.9 KB
/
Copy pathstrategy.py
File metadata and controls
1096 lines (907 loc) · 53.9 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
32
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
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
"""
交易策略模块，实现具体的交易策略逻辑
优化版本：统一止盈止损逻辑，优先处理止损，支持模拟交易
"""
import time
import threading
import logging
from datetime import datetime
import pandas as pd
import numpy as np
import config
from logger import get_logger, log_throttled, reset_log_throttle
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
# 导入卖出监控器 (容错处理)
try:
from sell_monitor import record_sell_attempt
SELL_MONITOR_ENABLED = True
except ImportError:
SELL_MONITOR_ENABLED = False
def record_sell_attempt(*args, **kwargs):
pass # 空函数
# 获取logger
logger = get_logger("strategy")
class TradingStrategy:
"""交易策略类，实现各种交易策略"""
SIGNAL_EXECUTION_SUCCESS = "success"
SIGNAL_EXECUTION_BLOCKED = "blocked"
SIGNAL_EXECUTION_FAILED = "failed"
def __init__(self):
"""初始化交易策略"""
self.data_manager = get_data_manager()
self.indicator_calculator = get_indicator_calculator()
self.position_manager = get_position_manager()
self.trading_executor = get_trading_executor()
# 策略运行线程
self.strategy_thread = None
self.stop_flag = False
# 防止频繁交易的冷却时间记录
self.last_trade_time = {}
# 🔒 线程安全：添加锁保护共享数据 (修复C1)
self.signal_lock = threading.Lock()
# 已处理的止盈止损信号记录
self.processed_signals = set()
# 添加这行 - 重试计数器
self.retry_counts = {}
# 信号去重集合所属交易日，跨日时清空，避免无人值守长跑内存单调增长
self.signals_date = datetime.now().strftime('%Y%m%d')
# MACD卖出开关关闭时的"已告知"记录，仅用于日志降噪
# 与 processed_signals 分离：开关盘中改为True时当日信号仍可立即执行
self.macd_sell_notified = set()
# ===== 旧的网格交易方法已废弃，请使用GridTradingManager =====
# init_grid_trading(), execute_grid_trading()
# 已被grid_trading_manager.py中的GridTradingManager替代
# ========== 新增：统一的止盈止损执行逻辑 ==========
def execute_trading_signal_direct(self, stock_code, signal_type, signal_info):
"""直接执行指定的交易信号"""
try:
# 🔑 添加统一信号验证
is_valid, validation_status, validation_reason = self.position_manager.validate_trading_signal(
stock_code, signal_type, signal_info, return_reason=True
)
if not is_valid:
if validation_status == "blocked":
# 阻断是持续性状态（T+1 冻结、委托在途），每轮都会命中，按股票+信号节流
log_throttled(
logger, logging.WARNING,
f"signal_blocked:{stock_code}:{signal_type}",
f"[信号阻断] {stock_code} {signal_type} 暂不执行，"
f"原因={validation_reason}，等待委托/持仓同步恢复后自动重试"
)
return self.SIGNAL_EXECUTION_BLOCKED
logger.warning(
f"[信号拒绝] {stock_code} {signal_type} 验证失败，"
f"原因={validation_reason}"
)
return self.SIGNAL_EXECUTION_FAILED
reset_log_throttle(f"signal_blocked:{stock_code}:{signal_type}")
if signal_type == 'stop_loss':
success = self._execute_stop_loss_signal(stock_code, signal_info)
elif signal_type == 'take_profit_half':
success = self._execute_take_profit_half_signal(stock_code, signal_info)
elif signal_type == 'take_profit_full':
success = self._execute_take_profit_full_signal(stock_code, signal_info)
else:
logger.warning(f"未知的信号类型: {signal_type}")
return self.SIGNAL_EXECUTION_FAILED
return self.SIGNAL_EXECUTION_SUCCESS if success else self.SIGNAL_EXECUTION_FAILED
except Exception as e:
logger.error(f"执行 {stock_code} 的 {signal_type} 信号时出错: {str(e)}")
return self.SIGNAL_EXECUTION_FAILED
def execute_add_position_strategy(self, stock_code, add_position_info):
"""
执行补仓策略
参数:
stock_code (str): 股票代码
add_position_info (dict): 补仓信号详细信息
返回:
bool: 是否执行成功
"""
try:
if self.position_manager._has_tracked_pending_order(stock_code):
logger.warning(f"[待委托拦截] {stock_code} 已有跟踪中的委托，跳过本次补仓")
return False
if (not getattr(config, 'ENABLE_SIMULATION_MODE', True)
and self.position_manager._has_pending_orders(stock_code)):
logger.warning(f"[待委托拦截] {stock_code} QMT存在活跃委托，跳过本次补仓")
return False
# 最终持仓限制检查（防止时差导致的超限）
position = self.position_manager.get_position(stock_code)
if position:
current_value = float(position.get('market_value', 0))
add_amount = add_position_info['add_amount']
if current_value + add_amount > config.MAX_POSITION_VALUE:
logger.warning(f"{stock_code} 补仓被拒绝: 当前市值{current_value:.2f} + 补仓{add_amount:.2f} = {current_value + add_amount:.2f} > 限制{config.MAX_POSITION_VALUE:.2f}")
return False
# 冷却期检查
cool_key = f"add_position_{stock_code}"
if cool_key in getattr(self, 'last_trade_time', {}):
last_time = self.last_trade_time[cool_key]
if (datetime.now() - last_time).total_seconds() < 120: # 2分钟冷却期
logger.debug(f"{stock_code} 补仓信号在冷却期内，跳过")
return False
add_amount = add_position_info['add_amount']
current_price = add_position_info['current_price']
logger.info(f"执行 {stock_code} 补仓策略，补仓金额: {add_amount:.2f}, 当前价格: {current_price:.2f}")
# 检查是否为模拟交易模式
if hasattr(config, 'ENABLE_SIMULATION_MODE') and config.ENABLE_SIMULATION_MODE:
# 模拟交易：计算买入数量
volume = int(add_amount // current_price / 100) * 100 # 向下取整到100的倍数
if volume < 100:
logger.warning(f"{stock_code} 计算的补仓数量过小: {volume}，跳过")
return False
# 模拟买入
success = self.position_manager.simulate_buy_position(
stock_code=stock_code,
buy_volume=volume,
buy_price=current_price
)
if success:
logger.info(f"[模拟交易] {stock_code} 补仓执行完成，数量: {volume}")
if not hasattr(self, 'last_trade_time'):
self.last_trade_time = {}
self.last_trade_time[cool_key] = datetime.now()
logger.info(f"{stock_code} 补仓成功，设置2分钟冷却期")
return True
else:
# 实盘交易：调用交易接口
logger.info(f"[实盘交易] {stock_code} 补仓信号已识别，执行实盘补仓")
# 使用金额买入方式
order_id = self.trading_executor.buy_stock(
stock_code,
amount=add_amount,
price_type=5,
strategy='add_position',
signal_type='add_position',
signal_info={
'current_price': current_price,
'add_amount': add_amount,
'order_side': 'BUY',
}
)
if not hasattr(self, 'last_trade_time'):
self.last_trade_time = {}
self.last_trade_time[cool_key] = datetime.now()
logger.info(f"{stock_code} 补仓成功，设置2分钟冷却期")
return order_id is not None
return False
except Exception as e:
logger.error(f"执行 {stock_code} 补仓策略时出错: {str(e)}")
return False
# def execute_trading_signal(self, stock_code):
# """
# 执行统一的交易信号处理 - 优化版本
# 参数:
# stock_code (str): 股票代码
# 返回:
# bool: 是否执行了交易操作
# """
# try:
# # 使用统一的信号检查函数
# signal_type, signal_info = self.position_manager.check_trading_signals(stock_code)
# if not signal_type:
# return False
# # 检查是否已处理过该信号（防重复处理）
# signal_key = f"{signal_type}_{stock_code}_{datetime.now().strftime('%Y%m%d_%H')}"
# if signal_key in self.processed_signals:
# logger.debug(f"{stock_code} {signal_type} 信号已处理，跳过")
# return False
# logger.info(f"处理 {stock_code} 的 {signal_type} 信号")
# # 根据信号类型执行相应操作
# success = False
# if signal_type == 'stop_loss':
# success = self._execute_stop_loss_signal(stock_code, signal_info)
# elif signal_type == 'take_profit_half':
# success = self._execute_take_profit_half_signal(stock_code, signal_info)
# elif signal_type == 'take_profit_full':
# success = self._execute_take_profit_full_signal(stock_code, signal_info)
# if success:
# # 记录已处理信号
# self.processed_signals.add(signal_key)
# logger.info(f"{stock_code} {signal_type} 信号处理成功")
# return success
# except Exception as e:
# logger.error(f"执行 {stock_code} 的交易信号时出错: {str(e)}")
# return False
def _execute_stop_loss_signal(self, stock_code, signal_info):
"""
执行止损信号
参数:
stock_code (str): 股票代码
signal_info (dict): 信号详细信息
返回:
bool: 是否执行成功
"""
try:
# ✅ 修复C2: 删除重复验证，信号验证已在execute_trading_signal_direct()中完成
volume = signal_info['volume']
current_price = signal_info['current_price']
logger.warning(f"执行 {stock_code} 止损操作，数量: {volume}, 当前价格: {current_price:.2f}")
# 检查是否为模拟交易模式
if hasattr(config, 'ENABLE_SIMULATION_MODE') and config.ENABLE_SIMULATION_MODE:
# 模拟交易：调用优化后的模拟卖出方法
success = self.position_manager.simulate_sell_position(
stock_code=stock_code,
sell_volume=volume,
sell_price=current_price,
sell_type='full'
)
if success:
logger.warning(f"[模拟交易] {stock_code} 止损执行完成，持仓已清零")
return success
else:
# 实盘交易：调用交易接口
logger.warning(f"[实盘交易] {stock_code} 止损信号已识别，执行实盘交易stop_loss")
# 实盘交易功能
# 🔑 新增：传递信号信息用于委托单跟踪
order_id = self.trading_executor.sell_stock(
stock_code, volume, price_type=5, strategy='stop_loss',
signal_type='stop_loss',
signal_info={
'current_price': current_price,
'cost_price': signal_info.get('cost_price', 0),
'volume': volume,
'loss_ratio': signal_info.get('loss_ratio', 0)
}
)
return order_id is not None
return False # 暂时返回False，表示未执行实盘交易
except Exception as e:
logger.error(f"执行 {stock_code} 止损信号时出错: {str(e)}")
return False
def _execute_take_profit_half_signal(self, stock_code, signal_info):
"""
执行首次止盈信号（卖出半仓）
参数:
stock_code (str): 股票代码
signal_info (dict): 信号详细信息
返回:
bool: 是否执行成功
"""
try:
# ✅ 修复C2: 删除重复验证，信号验证已在execute_trading_signal_direct()中完成
total_volume = signal_info['volume']
current_price = signal_info['current_price']
sell_ratio = signal_info['sell_ratio']
breakout_highest_price = signal_info.get('breakout_highest_price', 0)
pullback_ratio = signal_info.get('pullback_ratio', 0)
# 计算卖出数量
sell_volume = int(total_volume * sell_ratio / 100) * 100
sell_volume = max(sell_volume, 100) # 至少100股
logger.info(f"执行 {stock_code} 首次止盈，卖出半仓，数量: {sell_volume}, 价格: {current_price:.2f}")
if breakout_highest_price > 0:
logger.info(f" - 突破后最高价: {breakout_highest_price:.2f}, 回撤幅度: {pullback_ratio:.2%}")
# 检查是否为模拟交易模式
if hasattr(config, 'ENABLE_SIMULATION_MODE') and config.ENABLE_SIMULATION_MODE:
# 模拟交易：调用优化后的模拟卖出方法
success = self.position_manager.simulate_sell_position(
stock_code=stock_code,
sell_volume=sell_volume,
sell_price=current_price,
sell_type='partial'
)
if success:
# 验证执行结果
updated_position = self.position_manager.get_position(stock_code)
if updated_position and updated_position.get('profit_triggered'):
logger.info(f"[验证成功] {stock_code} 首次止盈执行完成并已标记")
return True
else:
logger.error(f"[验证失败] {stock_code} 首次止盈执行后状态异常")
return False
return success
else:
# 实盘交易：调用交易接口
logger.info(f"[实盘交易] {stock_code} 首次止盈信号已识别，执行实盘卖出交易take_profit_half")
# 实盘交易
# 🔑 新增：传递信号信息用于委托单跟踪
order_id = self.trading_executor.sell_stock(
stock_code, sell_volume, price_type=5, strategy='auto_partial',
signal_type='take_profit_half',
signal_info={
'current_price': current_price,
'cost_price': signal_info.get('cost_price', 0),
'volume': sell_volume,
'sell_ratio': sell_ratio,
'breakout_highest_price': breakout_highest_price,
'pullback_ratio': pullback_ratio
}
)
if order_id:
logger.info(f"[实盘交易] {stock_code} 首次止盈卖出委托已下达，委托号: {order_id}")
logger.info(f"[状态标记] {stock_code} 等待成交回报确认后再标记profit_triggered=True")
return True
else:
logger.error(f"[E_ORDER_SELL_101] {s
1798
tock_code} 首次止盈卖出委托下达失败，原因: trading_executor返回None (可能是ENABLE_ALLOW_SELL=False、持仓不足或QMT连接异常)，本次信号已标记为未处理，下次循环将自动重试")
return False
except Exception as e:
logger.error(f"执行 {stock_code} 首次止盈信号时出错: {str(e)}")
return False
def _execute_take_profit_full_signal(self, stock_code, signal_info):
"""
执行动态止盈信号（卖出剩余仓位）
参数:
stock_code (str): 股票代码
signal_info (dict): 信号详细信息
返回:
bool: 是否执行成功
"""
try:
# ✅ 修复C2: 删除重复验证，信号验证已在execute_trading_signal_direct()中完成
volume = signal_info['volume']
current_price = signal_info['current_price']
dynamic_take_profit_price = signal_info['dynamic_take_profit_price']
logger.info(f"执行 {stock_code} 动态止盈，卖出剩余仓位，数量: {volume}, "
f"当前价格: {current_price:.2f}, 止盈位: {dynamic_take_profit_price:.2f}")
# 检查是否为模拟交易模式
if hasattr(config, 'ENABLE_SIMULATION_MODE') and config.ENABLE_SIMULATION_MODE:
# 模拟交易：直接调整持仓
success = self.position_manager.simulate_sell_position(
stock_code=stock_code,
sell_volume=volume,
sell_price=current_price,
sell_type='full'
)
if success:
logger.info(f"[模拟交易] {stock_code} 动态止盈执行完成，持仓已清零")
return success
else:
# 实盘交易：调用交易接口
logger.info(f"[实盘交易] {stock_code} 动态止盈信号已识别，执行实盘卖出交易take_profit_full")
# 实盘交易
# 🔑 新增：传递信号信息用于委托单跟踪
order_id = self.trading_executor.sell_stock(
stock_code, volume, price_type=5, strategy='auto_full',
signal_type='take_profit_full',
signal_info={
'current_price': current_price,
'cost_price': signal_info.get('cost_price', 0),
'volume': volume,
'dynamic_take_profit_price': dynamic_take_profit_price,
'highest_price': signal_info.get('highest_price', 0)
}
)
if order_id:
logger.info(f"[实盘交易] {stock_code} 止盈全仓卖出委托已下达，委托号: {order_id}")
return True
else:
logger.error(f"[E_ORDER_SELL_102] {stock_code} 全仓止盈卖出委托下达失败，原因: trading_executor返回None (可能是ENABLE_ALLOW_SELL=False、持仓不足或QMT连接异常)，本次信号保留，将在下个策略循环重试(最多3次/分钟窗口)")
return False
return False # 暂时返回False，表示未执行实盘交易
except Exception as e:
logger.error(f"执行 {stock_code} 动态全仓止盈信号时出错: {str(e)}")
return False
# ========== 向后兼容的旧版本接口 ==========
# def execute_stop_loss(self, stock_code):
# """
# 执行止损策略 - 向后兼容接口
# 参数:
# stock_code (str): 股票代码
# 返回:
# bool: 是否执行成功
# """
# try:
# # 使用新的统一信号检查
# signal_type, signal_info = self.position_manager.check_trading_signals(stock_code)
# if signal_type == 'stop_loss':
# # 检查是否已处理过该信号
# signal_key = f"stop_loss_{stock_code}_{datetime.now().strftime('%Y%m%d')}"
# if signal_key in self.processed_signals:
# logger.debug(f"{stock_code} 止损信号已处理，跳过")
# return False
# success = self._execute_stop_loss_signal(stock_code, signal_info)
# if success:
# self.processed_signals.add(signal_key)
# return success
# return False
# except Exception as e:
# logger.error(f"执行 {stock_code} 的止损策略时出错: {str(e)}")
# return False
# def execute_dynamic_take_profit(self, stock_code):
# """
# 执行动态止盈策略 - 向后兼容接口
# 参数:
# stock_code (str): 股票代码
# 返回:
# bool: 是否执行成功
# """
# try:
# # 使用新的统一信号检查
# signal_type, signal_info = self.position_manager.check_trading_signals(stock_code)
# if signal_type in ['take_profit_half', 'take_profit_full']:
# # 检查是否已处理过该信号
# signal_key = f"take_profit_{stock_code}_{signal_type}_{datetime.now().strftime('%Y%m%d')}"
# if signal_key in self.processed_signals:
# logger.debug(f"{stock_code} {signal_type} 止盈信号已处理，跳过")
# return False
# success = False
# if signal_type == 'take_profit_half':
# success = self._execute_take_profit_half_signal(stock_code, signal_info)
# elif signal_type == 'take_profit_full':
# success = self._execute_take_profit_full_signal(stock_code, signal_info)
# if success:
# self.processed_signals.add(signal_key)
# return success
# return False
# except Exception as e:
# logger.error(f"执行 {stock_code} 的动态止盈策略时出错: {str(e)}")
# return False
def execute_buy_strategy(self, stock_code, buy_signal=None):
"""
执行买入策略
参数:
stock_code (str): 股票代码
buy_signal (bool): 外部传入的信号结果，避免重复调用 check_buy_signal；为 None 时内部自行检测
返回:
bool: 是否执行成功
"""
try:
# 优先使用外部传入的信号，避免重复调用 check_buy_signal
if buy_signal is None:
buy_signal = self.indicator_calculator.check_buy_signal(stock_code)
if buy_signal:
# 检查是否已处理过该信号
signal_key = f"buy_{stock_code}_{datetime.now().strftime('%Y%m%d')}"
if self._is_signal_processed(signal_key):
logger.debug(f"{stock_code} 买入信号已处理，跳过")
return False
# 检查是否已有持仓
position = self.position_manager.get_position(stock_code)
# 确定买入金额
if position:
# 已有持仓，检查是否达到补仓条件
current_price = position['current_price']
cost_price = position['cost_price']
current_value = position['market_value']
# 🔑 注意: execute_buy_strategy()仅处理技术指标买入信号的首次建仓
# 补仓策略已由position_manager.check_add_position_signal()独立处理
self._mark_signal_processed(signal_key) # 日内去重，防止每次循环都打印
logger.info(f"{stock_code} 已有持仓，技术指标买入信号不触发补仓（补仓由独立策略处理）")
return False
else:
# 新建仓，使用POSITION_UNIT作为首次建仓金额
buy_amount = config.POSITION_UNIT
logger.info(f"执行 {stock_code} 首次建仓，金额: {buy_amount:.2f}")
# 执行买入：与动态止盈/止损一致走对手价模式，买单由执行器取卖三价
order_id = self.trading_executor.buy_stock(stock_code, amount=buy_amount, price_type=5)
if order_id:
# 记录已处理信号
self._mark_signal_processed(signal_key)
# 旧的网格交易初始化已废弃，请使用GridTradingManager
return True
return False
except Exception as e:
logger.error(f"执行 {stock_code} 的买入策略时出错: {str(e)}")
return False
def execute_sell_strategy(self, stock_code, sell_signal=None):
"""
执行卖出策略
参数:
stock_code (str): 股票代码
sell_signal (bool): 外部传入的信号结果，避免重复调用 check_sell_signal；为 None 时内部自行检测
返回:
bool: 是否执行成功
"""
try:
# 优先使用外部传入的信号，避免重复调用 check_sell_signal
if sell_signal is None:
sell_signal = self.indicator_calculator.check_sell_signal(stock_code)
if sell_signal:
# 检查是否已处理过该信号
signal_key = f"sell_{stock_code}_{datetime.now().strftime('%Y%m%d')}"
if self._is_signal_processed(signal_key):
logger.debug(f"{stock_code} 卖出信号已处理，跳过")
return False
# MACD 技术指标卖出开关：默认关闭，满足条件仅记录信号，不执行真实卖出
# 降噪记录写入独立的 macd_sell_notified，不污染 processed_signals，
# 使得开关盘中改为 True 后当日信号无需重启进程即可立即生效
if not config.ENABLE_MACD_SELL:
with self.signal_lock:
already_notified = signal_key in self.macd_sell_notified
self.macd_sell_notified.add(signal_key)
if not already_notified:
logger.info(f"{stock_code} 检测到MACD技术指标卖出信号，但 ENABLE_MACD_SELL=False，仅记录信号不执行卖出")
return False
# 获取持仓
position = self.position_manager.get_position(stock_code)
if not position:
# 无持仓是暂时状态（盘中可能买入），不写 processed_signals 以免毒化当日信号，
# 仅用降噪集合防止每轮重复打印
with self.signal_lock:
already_warned = signal_key in self.macd_sell_notified
self.macd_sell_notified.add(signal_key)
if not already_warned:
logger.warning(f"未持有 {stock_code}，无法执行卖出策略")
return False
# 与止盈止损信号口径一致：使用可用持仓(available)下单，避免 T+1 冻结股份触发拒单
volume = int(position.get('available') or 0)
if volume <= 0:
logger.warning(
f"{stock_code} 可用持仓为0(总持仓={position.get('volume', 0)})，"
f"跳过MACD卖出，等待股份解冻后重试"
)
return False
# 执行卖出（price_type=5=LATEST_PRICE；旧值0为非法枚举，QMT客户端会本地拒绝且不产生废单）
logger.info(f"执行 {stock_code} 卖出策略，数量: {volume}")
order_id = self.trading_executor.sell_stock(stock_code, volume, price_type=5)
if order_id:
# 记录已处理信号
self._mark_signal_processed(signal_key)
return True
return False
except Exception as e:
logger.error(f"执行 {stock_code} 的卖出策略时出错: {str(e)}")
return False
def check_and_execute_strategies(self, stock_code):
"""
检查并执行所有交易策略 - 修复版本
策略检测始终运行，但交易执行依赖ENABLE_AUTO_TRADING
修复说明:
- 调整信号处理优先级: 止损 > 止盈 > 补仓 > 其他
- 止损作为最高优先级,确保风控底线
- 补仓前检查是否有止损信号,避免冲突
"""
try:
if not config.is_global_monitor_enabled():
logger.debug(f"全局自动操作总开关关闭，跳过 {stock_code} 自动策略执行")
return
# 添加调试日志
logger.debug(f"开始检查 {stock_code} 的交易策略，自动交易状态: {config.ENABLE_AUTO_TRADING}")
# 更新数据（始终执行）
self.data_manager.update_stock_data(stock_code)
self.indicator_calculator.calculate_all_indicators(stock_code)
# ========== 🔑 动态优先级信号处理 - 根据配置参数自动调整执行顺序 ==========
# 获取动态优先级信息
priority_info = config.determine_stop_loss_add_position_priority()
priority_mode = priority_info['priority']
scenario = priority_info['scenario']
# 场景A: 补仓优先 (补仓阈值 < 止损阈值, 例如补仓5% < 止损7%)
# 执行顺序: 止盈 → 补仓 → 止损
if priority_mode == 'add_position_first':
logger.debug(f"【场景{scenario}】使用补仓优先策略: 止盈 → 补仓 → 止损")
# 1️⃣ 止盈信号处理（第一优先级）
if config.ENABLE_DYNAMIC_STOP_PROFIT:
pending_signals = self.position_manager.get_pending_signals()
if stock_code in pending_signals:
signal_data = pending_signals[stock_code]
signal_type = signal_data['type']
signal_info = signal_data['info']
if signal_type in ['take_profit_half', 'take_profit_full']:
logger.info(f"{stock_code} 处理待执行的{signal_type}信号")
retry_key = f"{signal_type}_{stock_code}_{datetime.now().strftime('%Y%m%d_%H%M')}"
# 🔒 线程安全：使用锁保护retry_counts访问 (修复C1)
with self.signal_lock:
retry_count = self.retry_counts.get(retry_key, 0)
if retry_count >= 3:
logger.warning(f"{stock_code} {signal_type}信号重试次数已达上限")
self.position_manager.mark_signal_processed(stock_code)
return
if config.ENABLE_AUTO_TRADING:
result = self.execute_trading_signal_direct(stock_code, signal_type, signal_info)
if result == self.SIGNAL_EXECUTION_SUCCESS:
self.position_manager.mark_signal_processed(stock_code)
# 🔒 线程安全：使用锁保护retry_counts访问 (修复C1)
with self.signal_lock:
self.retry_counts.pop(retry_key, None)
logger.info(f"{stock_code} {signal_type}信号执行成功")
return # 止盈执行成功后直接返回
elif result == self.SIGNAL_EXECUTION_BLOCKED:
logger.debug(f"{stock_code} {signal_type}信号被委托/同步状态阻断，保留信号等待自动重试")
return
else:
# 🔒 线程安全：使用锁保护retry_counts访问 (修复C1)
with self.signal_lock:
self.retry_counts[retry_key] = retry_count + 1
if retry_count + 1 >= 3:
logger.error(f"🚨 {stock_code} {signal_type}信号重试{retry_count + 1}次仍失败，立即清除")
self.position_manager.mark_signal_processed(stock_code)
self.retry_counts.pop(retry_key, None)
else:
logger.info(f"{stock_code} 检测到{signal_type}信号，但自动交易已关闭")
self.position_manager.mark_signal_processed(stock_code)
# 2️⃣ 补仓信号处理（第二优先级）
add_position_signal, add_position_info = self.position_manager.check_add_position_signal(stock_code)
if add_position_signal == 'add_position':
logger.info(f"✅ 【场景{scenario}】{stock_code} 检测到补仓信号")
if config.ENABLE_AUTO_TRADING:
if self.execute_add_position_strategy(stock_code, add_position_info):
logger.info(f"{stock_code} 执行补仓策略成功")
return # 补仓执行后直接返回
else:
logger.info(f"{stock_code} 检测到补仓信号，但自动交易已关闭")
# 3️⃣ 止损信号处理（第三优先级 - 仅在仓位已满时触发）
if config.ENABLE_DYNAMIC_STOP_PROFIT:
pending_signals = self.position_manager.get_pending_signals()
if stock_code in pending_signals:
signal_data = pending_signals[stock_code]
signal_type = signal_data['type']
signal_info = signal_data['info']
if signal_type == 'stop_loss':
logger.warning(f"⚠️ 【场景{scenario}】{stock_code} 检测到止损信号(仓位已满)")
if config.ENABLE_AUTO_TRADING:
result = self.execute_trading_signal_direct(stock_code, signal_type, signal_info)
if result == self.SIGNAL_EXECUTION_SUCCESS:
self.position_manager.mark_signal_processed(stock_code)
logger.warning(f"✅ {stock_code} 止损信号执行成功")
return
elif result == self.SIGNAL_EXECUTION_BLOCKED:
logger.debug(f"{stock_code} 止损信号被委托/同步状态阻断，保留信号等待自动重试")
return
else:
logger.error(f"❌ {stock_code} 止损信号执行失败")
return
else:
logger.warning(f"{stock_code} 检测到止损信号，但自动交易已关闭")
self.position_manager.mark_signal_processed(stock_code)
return
# 场景B: 止损优先 (止损阈值 <= 补仓阈值, 例如止损5% <= 补仓7%)
# 执行顺序: 止损 → 止盈 → (永不补仓)
elif priority_mode == 'stop_loss_first':
logger.debug(f"【场景{scenario}】使用止损优先策略: 止损 → 止盈 → (永不补仓)")
# 1️⃣ 止损信号处理（最高优先级）
if config.ENABLE_DYNAMIC_STOP_PROFIT:
pending_signals = self.position_manager.get_pending_signals()
if stock_code in pending_signals:
signal_data = pending_signals[stock_code]
signal_type = signal_data['type']
signal_info = signal_data['info']
if signal_type == 'stop_loss':
logger.warning(f"⚠️ 【场景{scenario}】{stock_code} 检测到止损信号(最高优先级)，立即处理")
if config.ENABLE_AUTO_TRADING:
result = self.execute_trading_signal_direct(stock_code, signal_type, signal_info)
if result == self.SIGNAL_EXECUTION_SUCCESS:
self.position_manager.mark_signal_processed(stock_code)
logger.warning(f"✅ {stock_code} 止损信号执行成功，跳过其他策略")
return # 止损执行后直接返回
elif result == self.SIGNAL_EXECUTION_BLOCKED:
logger.debug(f"{stock_code} 止损信号被委托/同步状态阻断，保留信号等待自动重试")
return
else:
logger.error(f"❌ {stock_code} 止损信号执行失败")
return
else:
logger.warning(f"{stock_code} 检测到止损信号，但自动交易已关闭")
self.position_manager.mark_signal_processed(stock_code)
return
# 2️⃣ 止盈信号处理（第二优先级）
if config.ENABLE_DYNAMIC_STOP_PROFIT:
pending_signals = self.position_manager.get_pending_signals()
if stock_code in pending_signals:
signal_data = pending_signals[stock_code]
signal_type = signal_data['type']
signal_info = signal_data['info']
if signal_type in ['take_profit_half', 'take_profit_full']:
logger.info(f"{stock_code} 处理待执行的{signal_type}信号")
retry_key = f"{signal_type}_{stock_code}_{datetime.now().strftime('%Y%m%d_%H%M')}"
# 🔒 线程安全：使用锁保护retry_counts访问 (修复C1)
with self.signal_lock:
retry_count = self.retry_counts.get(retry_key, 0)
if retry_count >= 3:
logger.warning(f"{stock_code} {signal_type}信号重试次数已达上限")
self.position_manager.mark_signal_processed(stock_code)
return
if config.ENABLE_AUTO_TRADING:
result = self.execute_trading_signal_direct(stock_code, signal_type, signal_info)
if result == self.SIGNAL_EXECUTION_SUCCESS:
self.position_manager.mark_signal_processed(stock_code)
# 🔒 线程安全：使用锁保护retry_counts访问 (修复C1)
with self.signal_lock:
self.retry_counts.pop(retry_key, None)
logger.info(f"{stock_code} {signal_type}信号执行成功")
return
elif result == self.SIGNAL_EXECUTION_BLOCKED:
logger.debug(f"{stock_code} {signal_type}信号被委托/同步状态阻断，保留信号等待自动重试")
return
else:
# 🔒 线程安全：使用锁保护retry_counts访问 (修复C1)
with self.signal_lock:
self.retry_counts[retry_key] = retry_count + 1
if retry_count + 1 >= 3:
logger.error(f"🚨 {stock_code} {signal_type}信号重试{retry_count + 1}次仍失败，立即清除")
self.position_manager.mark_signal_processed(stock_code)
self.retry_counts.pop(retry_key, None)
else:
logger.info(f"{stock_code} 检测到{signal_type}信号，但自动交易已关闭")
self.position_manager.mark_signal_processed(stock_code)
# 3️⃣ 补仓信号 - 在场景B中永远不会触发
# check_add_position_signal() 已在 position_manager 中拒绝补仓
logger.debug(f"【场景{scenario}】补仓功能已禁用(止损优先策略)")
# 4. 清理历史遗留网格信号。网格交易已由 GridTradingManager 独立检测和执行。
pending_signals = self.position_manager.get_pending_signals()
if stock_code in pending_signals:
signal_type = pending_signals[stock_code]['type']
if signal_type in ['grid_buy', 'grid_sell', 'grid_exit']:
logger.info(f"[GRID-STRATEGY] {stock_code} 清理遗留网格信号: {signal_type}")
self.position_manager.mark_signal_processed(stock_code)
return
# 5. 检查技术指标买入信号
buy_signal = self.indicator_calculator.check_buy_signal(stock_code)
if buy_signal:
signal_key = f"buy_{stock_code}_{datetime.now().strftime('%Y%m%d')}"
if self._is_signal_processed(signal_key):
logger.debug(f"{stock_code} 买入信号今日已处理，跳过")
else:
logger.info(f"{st
B2C9
ock_code} 检测到买入信号")
# 只有在启用自动交易时才执行；传入已有信号避免重复调用
if config.ENABLE_AUTO_TRADING:
if self.execute_buy_strategy(stock_code, buy_signal=buy_signal):
logger.info(f"{stock_code} 执行买入策略成功")
return
else:
# 开关关闭仅降噪，不写 processed_signals，
# 使开关盘中打开后当日信号仍可执行
with self.signal_lock:
already_notified = signal_key in self.macd_sell_notified
self.macd_sell_notified.add(signal_key)
if not already_notified:
logger.info(f"{stock_code} 检测到买入信号，但自动交易已关闭")
# 6. 检查技术指标卖出信号
sell_signal = self.indicator_calculator.check_sell_signal(stock_code)
if sell_signal:
sell_key = f"sell_{stock_code}_{datetime.now().strftime('%Y%m%d')}"
if self._is_signal_processed(sell_key):
logger.debug(f"{stock_code} 卖出信号今日已处理，跳过")
else:
logger.info(f"{stock_code} 检测到卖出信号")
# 只有在启用自动交易时才执行
if config.ENABLE_AUTO_TRADING:
if self.execute_sell_strategy(stock_code, sell_signal=sell_signal):
logger.info(f"{stock_code} 执行卖出策略成功")
return
else:
# 同上：开关关闭仅降噪，不毒化当日信号
with self.signal_lock:
already_notified = sell_key in self.macd_sell_notified
self.macd_sell_notified.add(sell_key)
if not already_notified:
logger.info(f"{stock_code} 检测到卖出信号，但自动交易已关闭")
logger.debug(f"{stock_code} 没有检测到交易信号")
except Exception as e:
logger.error(f"检查 {stock_code} 的交易策略时出错: {str(e)}")
def start_strategy_thread(self):
"""启动策略运行线程 - 始终启动，不依赖ENABLE_AUTO_TRADING"""
if self.strategy_thread and self.strategy_thread.is_alive():
logger.warning("策略线程已在运行")
return
self.stop_flag = False
self.strategy_thread = threading.Thread(target=self._strategy_loop)
self.strategy_thread.daemon = True
self.strategy_thread.start()
logger.info("策略线程已启动（独立于自动交易开关）")
def stop_strategy_thread(self):
"""停止策略运行线程"""
if self.strategy_thread and self.strategy_thread.is_alive():
self.stop_flag = True
self.strategy_thread.join(timeout=5)
logger.info("策略线程已停止")
def _is_signal_processed(self, signal_key):
"""线程安全地判断信号是否已处理。"""
with self.signal_lock:
return signal_key in self.processed_signals
def _mark_signal_processed(self, signal_key):
"""线程安全地标记信号为已处理。"""
with self.signal_lock:
self.processed_signals.add(signal_key)
def _rollover_signal_cache_if_new_day(self):
"""跨交易日清空信号去重集合。
processed_signals / macd_sell_notified 的 key 带日期后缀，原本只在 close()
时清空。系统按无人值守设计长期运行，不清理会导致集合单调增长（每股每日最多
新增数个 key），且历史 key 永不释放。
返回:
bool: 本次调用是否发生了跨日清理
"""
today = datetime.now().strftime('%Y%m%d')
with self.signal_lock:
if today == self.signals_date:
return False
cleared = len(self.processed_signals) + len(self.macd_sell_notified)
self.processed_signals.clear()
self.macd_sell_notified.clear()
self.retry_counts.clear()
self.signals_date = today
logger.info(f"跨交易日清理信号去重集合，共清理 {cleared} 条记录，新交易日: {today}")
return True
def _strategy_loop(self):
"""策略运行循环 - 修复版本: 优先处理所有持仓股票"""
while not self.stop_flag:
try:
# 跨日清理信号去重集合（无论是否交易时段都检查，确保隔夜必然触发）
self._rollover_signal_cache_if_new_day()
# 判断是否在交易时间
if config.is_trade_time():
if config.VERBOSE_LOOP_LOGGING or config.DEBUG:
logger.debug("开始执行交易策略")
# 🔑 修复: 优先处理所有持仓股票 (止盈止损信号优先级最高)
positions = self.position_manager.get_all_positions()
processed_stocks = set()
if positions is not None and not positions.empty:
logger.debug(f"处理 {len(positions)} 只持仓股票的信号")
# ✅ 修复: 直接提取股票代码列表进行迭代
stock_codes = positions['stock_code'].tolist()
for stock_code in stock_codes:
self.check_and_execute_strategies(stock_code)
processed_stocks.add(stock_code)
time.sleep(1)
# 再处理STOCK_POOL中的其他股票 (买入信号等)
for stock_code in config.STOCK_POOL:
if stock_code not in processed_stocks: