fix: stop and start cheduler
This commit is contained in:
parent
c6966a96e0
commit
c443d0368a
@ -13,7 +13,7 @@ import os
|
|||||||
from config import Config
|
from config import Config
|
||||||
from logger_config import get_logger
|
from logger_config import get_logger
|
||||||
from core.base_trader import BaseTrader
|
from core.base_trader import BaseTrader
|
||||||
from core.app_state import login, logout, is_real_mode, _trader_instance, logger as app_logger # 使用 app_state中的logger
|
from core.app_state import get_real_trader_manager, login, logout, is_real_mode, _trader_instance, logger as app_logger # 使用 app_state中的logger
|
||||||
|
|
||||||
# 如果需要独立的 logger,可以取消下面这行的注释
|
# 如果需要独立的 logger,可以取消下面这行的注释
|
||||||
# logger = get_logger("scheduler")
|
# logger = get_logger("scheduler")
|
||||||
@ -44,6 +44,7 @@ def login_on_trading_day() -> None:
|
|||||||
if BaseTrader.is_trading_date():
|
if BaseTrader.is_trading_date():
|
||||||
logger.info("今天是交易日,执行计划的登录操作")
|
logger.info("今天是交易日,执行计划的登录操作")
|
||||||
login() # 调用 app_state.login
|
login() # 调用 app_state.login
|
||||||
|
get_real_trader_manager().start_scheduler()
|
||||||
else:
|
else:
|
||||||
logger.info("今天不是交易日,跳过计划的登录操作")
|
logger.info("今天不是交易日,跳过计划的登录操作")
|
||||||
|
|
||||||
@ -51,6 +52,7 @@ def logout_on_trading_day() -> None:
|
|||||||
"""仅在交易日执行登出操作。"""
|
"""仅在交易日执行登出操作。"""
|
||||||
if BaseTrader.is_trading_date():
|
if BaseTrader.is_trading_date():
|
||||||
logger.info("今天是交易日,执行计划的登出操作")
|
logger.info("今天是交易日,执行计划的登出操作")
|
||||||
|
get_real_trader_manager().stop_scheduler()
|
||||||
logout() # 调用 app_state.logout
|
logout() # 调用 app_state.logout
|
||||||
else:
|
else:
|
||||||
logger.info("今天不是交易日,跳过计划的登出操作")
|
logger.info("今天不是交易日,跳过计划的登出操作")
|
||||||
|
@ -47,12 +47,63 @@ class RealTraderManager:
|
|||||||
# 初始化锁
|
# 初始化锁
|
||||||
self._lock = threading.Lock()
|
self._lock = threading.Lock()
|
||||||
|
|
||||||
|
# 添加停止事件
|
||||||
|
self._stop_event = threading.Event()
|
||||||
|
|
||||||
|
self.scheduler_thread = None
|
||||||
|
|
||||||
# 启动调度器
|
# 启动调度器
|
||||||
self._start_scheduler()
|
self._start_scheduler()
|
||||||
|
|
||||||
logger.info("实盘交易管理器初始化完成")
|
logger.info("实盘交易管理器初始化完成")
|
||||||
|
|
||||||
def _start_scheduler(self):
|
def stop_scheduler(self) -> None:
|
||||||
|
"""停止调度器线程
|
||||||
|
|
||||||
|
正确的停止流程:
|
||||||
|
1. 设置停止事件通知线程退出
|
||||||
|
2. 等待线程自然结束
|
||||||
|
3. 清理线程引用
|
||||||
|
"""
|
||||||
|
if self.scheduler_thread and self.scheduler_thread.is_alive():
|
||||||
|
logger.info("正在停止调度器...")
|
||||||
|
# 设置停止事件
|
||||||
|
self._stop_event.set()
|
||||||
|
|
||||||
|
# 等待线程结束,设置超时避免无限等待
|
||||||
|
try:
|
||||||
|
self.scheduler_thread.join(timeout=5.0)
|
||||||
|
if self.scheduler_thread.is_alive():
|
||||||
|
logger.warning("调度器线程未能在超时时间内停止")
|
||||||
|
else:
|
||||||
|
logger.info("调度器已成功停止")
|
||||||
|
except Exception as e:
|
||||||
|
logger.error(f"等待调度器线程结束时发生异常: {e}")
|
||||||
|
|
||||||
|
self.scheduler_thread = None
|
||||||
|
else:
|
||||||
|
logger.info("调度器未运行或已停止")
|
||||||
|
|
||||||
|
def start_scheduler(self) -> None:
|
||||||
|
"""启动调度器(公共方法,支持重启)"""
|
||||||
|
if self.scheduler_thread and self.scheduler_thread.is_alive():
|
||||||
|
logger.warning("调度器已在运行,请先停止")
|
||||||
|
return
|
||||||
|
|
||||||
|
# 重置停止事件
|
||||||
|
self._stop_event.clear()
|
||||||
|
self._start_scheduler()
|
||||||
|
|
||||||
|
def __del__(self) -> None:
|
||||||
|
"""析构方法,确保资源清理"""
|
||||||
|
try:
|
||||||
|
self.stop_scheduler()
|
||||||
|
except Exception as e:
|
||||||
|
# 析构方法中的异常不应该向上传播
|
||||||
|
pass
|
||||||
|
|
||||||
|
def _start_scheduler(self) -> None:
|
||||||
|
"""内部方法:启动调度器线程"""
|
||||||
# 每日定时清理(增加配置校验)
|
# 每日定时清理(增加配置校验)
|
||||||
if hasattr(Config, "CLEAN_ORDERS_TIME"):
|
if hasattr(Config, "CLEAN_ORDERS_TIME"):
|
||||||
try:
|
try:
|
||||||
@ -92,19 +143,26 @@ class RealTraderManager:
|
|||||||
|
|
||||||
# 启动高精度调度线程
|
# 启动高精度调度线程
|
||||||
def run_scheduler():
|
def run_scheduler():
|
||||||
while True:
|
"""调度器线程主循环,响应停止事件"""
|
||||||
|
logger.info("调度器线程开始运行")
|
||||||
|
while not self._stop_event.is_set():
|
||||||
try:
|
try:
|
||||||
schedule.run_pending()
|
schedule.run_pending()
|
||||||
time.sleep(1) # 将休眠时间缩短至1秒提高精度
|
# 使用wait代替sleep,这样可以立即响应停止事件
|
||||||
|
self._stop_event.wait(timeout=1.0)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error(f"调度器异常: {e}", exc_info=True)
|
logger.error(f"调度器异常: {e}", exc_info=True)
|
||||||
time.sleep(10) # 发生错误时延长休眠避免日志风暴
|
# 发生错误时也要检查停止事件
|
||||||
|
if not self._stop_event.is_set():
|
||||||
|
self._stop_event.wait(timeout=10.0)
|
||||||
|
|
||||||
scheduler_thread = threading.Thread(
|
logger.info("调度器线程收到停止信号,正在退出")
|
||||||
|
|
||||||
|
self.scheduler_thread = threading.Thread(
|
||||||
target=run_scheduler, name="SchedulerThread"
|
target=run_scheduler, name="SchedulerThread"
|
||||||
)
|
)
|
||||||
scheduler_thread.daemon = True # 设为守护线程随主进程退出
|
self.scheduler_thread.daemon = True # 设为守护线程随主进程退出
|
||||||
scheduler_thread.start()
|
self.scheduler_thread.start()
|
||||||
logger.info("交易管理器调度器已启动")
|
logger.info("交易管理器调度器已启动")
|
||||||
|
|
||||||
def place_order(
|
def place_order(
|
||||||
|
Loading…
x
Reference in New Issue
Block a user