demand_server/app/scheduler/manager.py

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.interval import IntervalTrigger

from app.core.config import settings
from app.scheduler.jobs import (
    demand_pool_daily_strategy_alert_job,
    demand_pool_today_incremental_sync_job,
    heartbeat_job,
)

# 与 scheduler 一致。APScheduler 3.x 在 add_job 里传入 CronTrigger 实例时，若未显式指定
# timezone，会按 UTC 等默认时区解析 hour/minute，导致「配置的 9:40」实际在 UTC 触发，
# 北京时间观测为 17:40（+8 小时）。
_CRON_TZ = "Asia/Shanghai"

scheduler = AsyncIOScheduler(timezone=_CRON_TZ)


def setup_jobs() -> None:
    scheduler.add_job(
        heartbeat_job,
        trigger=IntervalTrigger(seconds=settings.scheduler_heartbeat_seconds),
        id="heartbeat_job",
        replace_existing=True,
    )
    if settings.demand_pool_hourly_sync_enabled:
        scheduler.add_job(
            demand_pool_today_incremental_sync_job,
            trigger=CronTrigger(
                minute=settings.demand_pool_hourly_sync_minute,
                timezone=_CRON_TZ,
            ),
            id="demand_pool_hourly_sync_job",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )
    if settings.demand_pool_daily_strategy_alert_enabled:
        scheduler.add_job(
            demand_pool_daily_strategy_alert_job,
            trigger=CronTrigger(
                hour=settings.demand_pool_daily_strategy_alert_hour,
                minute=settings.demand_pool_daily_strategy_alert_minute,
                timezone=_CRON_TZ,
            ),
            id="demand_pool_daily_strategy_alert_job",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )


def start_scheduler() -> None:
    if scheduler.running:
        return
    setup_jobs()
    scheduler.start()


def shutdown_scheduler() -> None:
    if not scheduler.running:
        return
    scheduler.shutdown(wait=False)