Repository navigation
Expand file tree
/
Copy pathstrategy.py
More file actions
823 lines (656 loc) · 35 KB
/
Copy pathstrategy.py
File metadata and controls
823 lines (656 loc) · 35 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
"""
交易策略模块，实现具体的交易策略逻辑
优化版本：统一止盈止损逻辑，优先处理止损，支持模拟交易
"""
import time
import threading
from datetime import datetime
import pandas as pd
import numpy as np
import config
from logger import get_logger
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
# 获取logger
logger = get_logger("strategy")
class TradingStrategy:
"""交易策略类，实现各种交易策略"""
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 = {}
# 已处理的止盈止损信号记录
self.processed_signals = set()
# 添加这行 - 重试计数器
self.retry_counts = {}
def init_grid_trading(self, stock_code):
"""
初始化网格交易
参数:
stock_code (str): 股票代码
返回:
bool: 是否初始化成功
"""
try:
if not config.ENABLE_GRID_TRADING:
logger.info(f"网格交易功能未启用，跳过 {stock_code} 的网格初始化")
return False
# 获取持仓信息
position = self.position_manager.get_position(stock_code)
if not position:
logger.warning(f"未持有 {stock_code}，无法初始化网格交易")
return False
# 获取最新行情
latest_quote = self.data_manager.get_latest_data(stock_code)
if not latest_quote:
logger.error(f"未能获取 {stock_code} 的最新行情，无法初始化网格交易")
return False
current_price = latest_quote.get('lastPrice')
position_volume = position['volume']
# 清除旧的网格记录
# 这里需要在position_manager中添加一个清除网格交易的方法，暂时跳过
# 创建网格
grid_count = min(config.GRID_MAX_LEVELS, 5) # 最多创建5个网格
grid_volume = int(position_volume * config.GRID_POSITION_RATIO / grid_count)
if grid_volume < 100:
logger.warning(f"{stock_code} 持仓量不足，无法创建有效的网格交易")
return False
for i in range(grid_count):
# 买入价格递减，卖出价格递增
buy_price = current_price * (1 - config.GRID_STEP_RATIO * (i + 1))
sell_price = current_price * (1 + config.GRID_STEP_RATIO * (i + 1))
# 创建网格交易
grid_id = self.position_manager.add_grid_trade(
stock_code, i + 1, buy_price, sell_price, grid_volume
)
if grid_id < 0:
logger.error(f"创建 {stock_code} 的网格交易记录失败")
return False
logger.info(f"初始化 {stock_code} 的网格交易成功，创建了 {grid_count} 个网格")
return True
except Exception as e:
logger.error(f"初始化 {stock_code} 的网格交易时出错: {str(e)}")
return False
def execute_grid_trading(self, stock_code):
"""
执行网格交易策略
参数:
stock_code (str): 股票代码
返回:
bool: 是否执行成功
"""
try:
if not config.ENABLE_GRID_TRADING:
return False
# 检查是否有网格交易信号
grid_signals = self.position_manager.check_grid_trade_signals(stock_code)
# 处理买入信号
for signal in grid_signals['buy_signals']:
grid_id = signal['grid_id']
price = signal['price']
volume = signal['volume']
# 检查同一网格是否已经在冷却期
cool_key = f"grid_buy_{stock_code}_{grid_id}"
if cool_key in self.last_trade_time:
last_time = self.last_trade_time[cool_key]
if (datetime.now() - last_time).total_seconds() < 300: # 5分钟冷却期
logger.debug(f"{stock_code} 网格 {grid_id} 买入信号在冷却期内，跳过")
continue
# 执行买入
logger.info(f"执行 {stock_code} 网格 {grid_id} 买入，价格: {price}, 数量: {volume}")
order_id = self.trading_executor.buy_stock(stock_code, volume, price, strategy='grid')
if order_id:
# 更新网格状态为活跃
self.position_manager.update_grid_trade_status(grid_id, 'ACTIVE')
# 记录交易时间
self.last_trade_time[cool_key] = datetime.now()
# 处理卖出信号
for signal in grid_signals['sell_signals']:
grid_id = signal['grid_id']
price = signal['price']
volume = signal['volume']
# 检查同一网格是否已经在冷却期
cool_key = f"grid_sell_{stock_code}_{grid_id}"
if cool_key in self.last_trade_time:
last_time = self.last_trade_time[cool_key]
if (datetime.now() - last_time).total_seconds() < 300: # 5分钟冷却期
logger.debug(f"{stock_code} 网格 {grid_id} 卖出信号在冷却期内，跳过")
continue
# 执行卖出
logger.info(f"执行 {stock_code} 网格 {grid_id} 卖出，价格: {price}, 数量: {volume}")
order_id = self.trading_executor.sell_stock(stock_code, volume, price, strategy='grid')
if order_id:
# 更新网格状态为完成
self.position_manager.update_grid_trade_status(grid_id, 'COMPLETED')
# 记录交易时间
self.last_trade_time[cool_key] = datetime.now()
return True
except Exception as e:
logger.error(f"执行 {stock_code} 的网格交易时出错: {str(e)}")
return False
# ========== 新增：统一的止盈止损执行逻辑 ==========
def execute_trading_signal_direct(self, stock_code, signal_type, signal_info):
"""直接执行指定的交易信号"""
try:
if signal_type == 'stop_loss':
return self._execute_stop_loss_signal(stock_code, signal_info)
elif signal_type == 'take_profit_half':
return self._execute_take_profit_half_signal(stock_code, signal_info)
elif signal_type == 'take_profit_full':
return self._execute_take_profit_full_signal(stock_code, signal_info)
else:
logger.warning(f"未知的信号类型: {signal_type}")
return False
except Exception as e:
logger.error(f"执行 {stock_code} 的 {signal_type} 信号时出错: {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:
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'
)
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:
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'
)
if order_id:
logger.info(f"[实盘交易] {stock_code} 首次止盈卖出委托已下达，委托号: {order_id}")
# 标记已触发首次止盈
mark_success = self.position_manager.mark_profit_triggered(stock_code)
if mark_success:
logger.info(f"[状态标记] {stock_code} profit_triggered已标记为True")
# 返回True表示整个止盈操作成功
return True
else:
logger.error(f"[状态标记] {stock_code} profit_triggered标记失败")
return False
else:
logger.error(f"[实盘交易] {stock_code} 首次止盈卖出委托下达失败")
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:
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'
)
if order_id:
logger.info(f"[实盘交易] {stock_code} 止盈全仓卖出委托已下达，委托号: {order_id}")
return True
else:
logger.error(f"[实盘交易] {stock_code} 全仓止盈卖出委托下达失败")
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):
"""
执行买入策略
参数:
stock_code (str): 股票代码
返回:
bool: 是否执行成功
"""
try:
# 检查是否有买入信号
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 signal_key in self.processed_signals:
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']
# 检查是否满足补仓格点要求
price_ratio = current_price / cost_price
# 寻找满足条件的补仓格点
buy_level = None
for i, level in enumerate(config.BUY_GRID_LEVELS):
if i > 0 and price_ratio <= level: # 不是第一格且价格比例小于等于格点比例
buy_level = i
break
if buy_level is None:
logger.info(f"{stock_code} 当前价格不满足补仓条件")
return False
# 检查是否达到最大持仓限制
if current_value >= config.MAX_POSITION_VALUE:
logger.info(f"{stock_code} 持仓已达到最大限制，不再补仓")
return False
# 确定补仓金额
buy_amount = config.POSITION_UNIT * config.BUY_AMOUNT_RATIO[buy_level]
logger.info(f"执行 {stock_code} 补仓策略，当前价格比例: {price_ratio:.2f}, 补仓格点: {buy_level}, 补仓金额: {buy_amount}")
else:
# 新建仓，使用第一个格点的金额
buy_amount = config.POSITION_UNIT * config.BUY_AMOUNT_RATIO[0]
logger.info(f"执行 {stock_code} 首次建仓，金额: {buy_amount}")
# 执行买入
order_id = self.trading_executor.buy_stock(stock_code, amount=buy_amount, price_type=0)
if order_id:
# 记录已处理信号
self.processed_signals.add(signal_key)
# 如果是新建仓，初始化网格交易
if not position and config.ENABLE_GRID_TRADING:
# 等待买入成交后再初始化网格
# 实际应用中应该通过回调函数处理
time.sleep(5) # 简单等待一下
self.init_grid_trading(stock_code)
return True
return False
except Exception as e:
logger.error(f"执行 {stock_code} 的买入策略时出错: {str(e)}")
return False
def execute_sell_strategy(self, stock_code):
"""
执行卖出策略
参数:
stock_code (str): 股票代码
返回:
bool: 是否执行成功
"""
try:
# 检查是否有卖出信号
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 signal_key in self.processed_signals:
logger.debug(f"{stock_code} 卖出信号已处理，跳过")
return False
# 获取持仓
position = self.position_manager.get_position(stock_code)
if not position:
logger.warning(f"未持有 {stock_code}，无法执行卖出策略")
return False
volume = position['volume']
# 执行卖出
logger.info(f"执行 {stock_code} 卖出策略，数量: {volume}")
order_id = self.trading_executor.sell_stock(stock_code, volume, price_type=0)
if order_id:
# 记录已处理信号
self.processed_signals.add(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:
# 添加调试日志
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)
# 1. 检查止盈止损信号（如果启用）
if config.ENABLE_DYNAMIC_STOP_PROFIT:
pending_signals = self.position_manager.get_pending_signals()
# 添加调试日志
logger.debug(f"{stock_code} 待处理信号: {list(pending_signals.keys())}")
if stock_code in pending_signals:
signal_data = pending_signals[stock_code]
signal_type = signal_data['type']
signal_info = signal_data['info']
logger.info(f"{stock_code} 处理待执行的{signal_type}信号")
# 检查是否已处理过该信号（防重复,每分钟3次）
retry_key = f"{signal_type}_{stock_code}_{datetime.now().strftime('%Y%m%d_%H%M')}"
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:
# 添加调试日志
logger.info(f"{stock_code} 开始执行{signal_type}信号，重试次数: {retry_count}")
success = self.execute_trading_signal_direct(stock_code, signal_type, signal_info)
if success:
self.position_manager.mark_signal_processed(stock_code)
self.retry_counts.pop(retry_key, None)
logger.info(f"{stock_code} {signal_type}信号执行成功")
else:
self.retry_counts[retry_key] = retry_count + 1
logger.warning(f"{stock_code} {signal_type}执行失败，重试次数: {retry_count + 1}")
else:
logger.info(f"{stock_code} 检测到{signal_type}信号，但自动交易已关闭")
self.position_manager.mark_signal_processed(stock_code)
else:
logger.debug(f"{stock_code} 当前无待处理信号")
# 2. 检查网格交易信号（如果启用）
if config.ENABLE_GRID_TRADING:
grid_signals = self.position_manager.check_grid_trade_signals(stock_code)
if grid_signals['buy_signals'] or grid_signals['sell_signals']:
logger.info(f"{stock_code} 检测到网格交易信号")
# 只有在启用自动交易时才执行
if config.ENABLE_AUTO_TRADING:
if self.execute_grid_trading(stock_code):
logger.info(f"{stock_code} 执行网格交易策略成功")
return
else:
logger.info(f"{stock_code} 检测到网格信号，但自动交易已关闭")
# 3. 检查技术指标买入信号
buy_signal = self.indicator_calculator.check_buy_signal(stock_code)
if buy_signal:
logger.info(f"{stock_code} 检测到买入信号")
# 只有在启用自动交易时才执行
if config.ENABLE_AUTO_TRADING:
if self.execute_buy_strategy(stock_code):
logger.info(f"{stock_code} 执行买入策略成功")
return
else:
logger.info(f"{stock_code} 检测到买入信号，但自动交易已关闭")
# 4. 检查技术指标卖出信号
sell_signal = self.indicator_calculator.check_sell_signal(stock_code)
if sell_signal:
logger.info(f"{stock_code} 检测到卖出信号")
# 只有在启用自动交易时才执行
if config.ENABLE_AUTO_TRADING:
if self.execute_sell_strategy(stock_code):
logger.info(f"{stock_code} 执行卖出策略成功")
return
else:
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 _strategy_loop(self):
"""策略运行循环"""
while not self.stop_flag:
try:
# 判断是否在交易时间
if config.is_trade_time():
logger.info("开始执行交易策略")
# 遍历股票池中的每只股票
for stock_code in config.STOCK_POOL:
# 检查并执行交易策略
self.check_and_execute_strategies(stock_code)
# 避免请求过于频繁
time.sleep(1)
logger.info("交易策略执行完成")
# 等待下一次策略执行
for _ in range(10): # 每10s执行一次策略
if self.stop_flag:
break
time.sleep(1)
except Exception as e:
logger.error(f"策略循环出错: {str(e)}")
time.sleep(60) # 出错后等待一分钟再继续
def manual_buy(self, stock_code, volume=None, price=None, amount=None):
"""
手动买入股票 - 不受ENABLE_AUTO_TRADING限制
"""
try:
# 手动交易不检查ENABLE_AUTO_TRADING，但要检查ENABLE_ALLOW_BUY
if not config.ENABLE_ALLOW_BUY:
logger.warning(f"系统当前不允许买入操作")
return None
# 根据交易模式选择策略标识
is_simulation = hasattr(config, 'ENABLE_SIMULATION_MODE') and config.ENABLE_SIMULATION_MODE
strategy = 'M_simu' if is_simulation else 'M_real'
order_id = self.trading_executor.buy_stock(
stock_code, volume, price, amount, strategy=strategy
)
if order_id:
logger.info(f"手动买入 {stock_code} 成功，委托号: {order_id}，模式: {'模拟' if is_simulation else '实盘'}")
return order_id
except Exception as e:
logger.error(f"手动买入 {stock_code} 时出错: {str(e)}")
return None
def manual_sell(self, stock_code, volume=None, price=None, ratio=None):
"""
手动卖出股票 - 不受ENABLE_AUTO_TRADING限制
"""
try:
# 手动交易不检查ENABLE_AUTO_TRADING，但要检查ENABLE_ALLOW_SELL
if not config.ENABLE_ALLOW_SELL:
logger.warning(f"系统当前不允许卖出操作")
return None
# 根据交易模式选择策略标识
is_simulation = hasattr(config, 'ENABLE_SIMULATION_MODE') and config.ENABLE_SIMULATION_MODE
strategy = 'manual_simu' if is_simulation else 'manual_real'
order_id = self.trading_executor.sell_stock(
stock_code, volume, price, ratio, strategy=strategy
)
if order_id:
logger.info(f"手动卖出 {stock_code} 成功，委托号: {order_id}，模式: {'模拟' if is_simulation else '实盘'}")
return order_id
except Exception as e:
logger.error(f"手动卖出 {stock_code} 时出错: {str(e)}")
return None
# 单例模式
_instance = None
def get_trading_strategy():
"""获取TradingStrategy单例"""
global _instance
if _instance is None:
_instance = TradingStrategy()
return _instance