From cacdb5ae248c518b3dd8c2b4633fddad1693f46a Mon Sep 17 00:00:00 2001 From: claude_dev Date: Tue, 7 Jul 2026 12:08:32 +0800 Subject: [PATCH] =?UTF-8?q?feat(trader):=20C-S3=20scheduler(APScheduler?= =?UTF-8?q?=E5=AE=9A=E6=97=B6+=E5=90=AF=E5=8A=A8=E6=81=A2=E5=A4=8Dlive=20j?= =?UTF-8?q?ob)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- sanguo_trader/scheduler.py | 64 ++++++++++++++++++++++++++++++++++ tests/trader/test_scheduler.py | 35 +++++++++++++++++++ 2 files changed, 99 insertions(+) create mode 100644 sanguo_trader/scheduler.py create mode 100644 tests/trader/test_scheduler.py diff --git a/sanguo_trader/scheduler.py b/sanguo_trader/scheduler.py new file mode 100644 index 0000000..ac43206 --- /dev/null +++ b/sanguo_trader/scheduler.py @@ -0,0 +1,64 @@ +"""模拟盘 C 实走调度(APScheduler 定时 + 启动恢复,spec §9.2)。 + +每日 20:30 触发 step(拉当日 bar + engine.step)。APScheduler 是 pip 依赖, +lazy import;本机无则单测 mock get_scheduler。 +""" +import logging +import sqlite3 + +logger = logging.getLogger(__name__) +_scheduler = None + + +def get_scheduler(): + global _scheduler + if _scheduler is None: + from apscheduler.schedulers.background import BackgroundScheduler + _scheduler = BackgroundScheduler() + return _scheduler + + +def start_scheduler(): + s = get_scheduler() + if not s.running: + s.start() + return s + + +def register_paper_job(account_id: int, hour: int = 20, minute: int = 30, + job_func=None) -> str: + """注册每日定时 job,返回 job_id(存 paper_accounts.scheduler_job_id)。""" + s = start_scheduler() + job_id = f"paper_{account_id}" + s.add_job( + job_func or _default_step, + "cron", # 字符串 trigger,避免函数内 import CronTrigger + hour=hour, minute=minute, + args=[account_id], id=job_id, replace_existing=True, + ) + return job_id + + +def remove_paper_job(account_id: int) -> None: + s = get_scheduler() + job_id = f"paper_{account_id}" + try: + s.remove_job(job_id) + except Exception: + logger.warning("remove_job %s not found", job_id) + + +def _default_step(account_id: int) -> None: + """部署时注入真 step(akshare 拉当日 + engine.step);默认 stub。""" + logger.info("paper account %s daily step (stub)", account_id) + + +def restore_live_jobs(db_path: str, job_func=None) -> int: + """容器启动:遍历 status=running & mode=live 重建 job(spec §9.2 H-3)。""" + with sqlite3.connect(db_path) as conn: + rows = conn.execute( + "SELECT id FROM paper_accounts WHERE status='running' AND mode='live'" + ).fetchall() + for (aid,) in rows: + register_paper_job(aid, job_func=job_func) + return len(rows) diff --git a/tests/trader/test_scheduler.py b/tests/trader/test_scheduler.py new file mode 100644 index 0000000..9c0e781 --- /dev/null +++ b/tests/trader/test_scheduler.py @@ -0,0 +1,35 @@ +"""scheduler 测试(mock APScheduler,spec §9.2)。""" +from unittest.mock import MagicMock + +from sanguo_trader import scheduler +from sanguo_trader.persistence import ( + init_db, save_account, update_account_status, +) + + +def test_register_paper_job(monkeypatch): + fake = MagicMock() + monkeypatch.setattr(scheduler, "start_scheduler", lambda: fake) + job_id = scheduler.register_paper_job(123, job_func=lambda a: None) + assert job_id == "paper_123" + fake.add_job.assert_called_once() + + +def test_remove_paper_job(monkeypatch): + fake = MagicMock() + monkeypatch.setattr(scheduler, "get_scheduler", lambda: fake) + scheduler.remove_paper_job(123) + fake.remove_job.assert_called_once_with("paper_123") + + +def test_restore_live_jobs_filters_running_live(monkeypatch, tmp_path): + db = str(tmp_path / "s.db") + init_db(db) + a_live = save_account(db, {"name": "t1", "mode": "live"}) + a_replay = save_account(db, {"name": "t2", "mode": "replay"}) + update_account_status(db, a_live, "running") + update_account_status(db, a_replay, "running") + fake = MagicMock() + monkeypatch.setattr(scheduler, "start_scheduler", lambda: fake) + n = scheduler.restore_live_jobs(db, job_func=lambda a: None) + assert n == 1 # 只 live & running