每日选股信号的自动化：systemd timer + 企业微信推送

远思为径，码动价值。欢迎来到远思码技术实验室的最新技术实践记录。

研究做得再漂亮，如果每天还要手动开 Jupyter 跑一遍 notebook，那它就永远只是个练习。这一篇把前面六篇串成一条无人值守的流水线：每个交易日盘后自动取数、算因子、生成信号，并把结果推到手机上。

一、整条链路

┌─────────┐   ┌──────────┐   ┌──────────┐   ┌──────────┐   ┌─────────┐
│ 取行情  │ → │ 算因子   │ → │ 模型打分 │ → │ 组合构建 │ → │ 推送+落库│
└─────────┘   └──────────┘   └──────────┘   └──────────┘   └─────────┘
     失败 → 重试 3 次 → 仍失败则告警 → 记录死信

设计原则只有三条：

幂等：同一天重跑一百次，结果一样，不会重复推送；
可见：每次运行都有日志、有退出码、有心跳；
可停：任何一个环节失败都不会污染已有信号。

二、先判断是不是交易日

节假日不跑，是这条流水线最重要的一条规则。systemd 的 OnCalendar 只认星期几，不认 A 股节假日，所以判断必须写在程序里。

# quant/calendar.py
from __future__ import annotations

import datetime as dt

import akshare as ak
import pandas as pd

_cache: pd.DataFrame | None = None


def trade_dates() -> list[dt.date]:
    """A 股历史交易日序列（含未来已公布日期）。首次调用约 1 秒，之后走缓存。"""
    global _cache
    if _cache is None:
        df = ak.tool_trade_date_hist_sina()      # 列: trade_date
        _cache = pd.to_datetime(df["trade_date"]).dt.date
    return list(_cache)


def is_trading_day(day: dt.date | None = None) -> bool:
    day = day or dt.date.today()
    return day in trade_dates()


def last_trading_day(day: dt.date | None = None) -> dt.date:
    """最近一个 <= day 的交易日（用于补跑）。"""
    day = day or dt.date.today()
    dates = [d for d in trade_dates() if d <= day]
    if not dates:
        raise RuntimeError(f"{day} 之前没有交易日数据，检查交易日历缓存")
    return dates[-1]

为什么不干脆只在 OnCalendar=Mon..Fri 上跑？因为调休。2026 年的春节、国庆前后都有"周六上班"的交易日，只在周一到周五跑会漏；反过来节假日又白跑。让程序判断交易日、让 timer 每天都触发一次，是最省心的组合。

三、主流程：一次运行做五件事

# quant/pipeline.py
from __future__ import annotations

import datetime as dt
import logging
import sys

import pandas as pd

from quant.calendar import is_trading_day, last_trading_day
from quant.config import CFG
from quant.db import save_signals
from quant.features import build_features
from quant.model import load_model, score_panel
from quant.notify import push_wecom
from quant.portfolio import build_portfolio

log = logging.getLogger("quant.pipeline")


def run_once(force: bool = False, day: dt.date | None = None) -> int:
    """返回进程退出码：0 成功/跳过，1 失败。"""
    day = day or last_trading_day()
    if not force and not is_trading_day(day):
        log.info("非交易日 %s，跳过", day)
        return 0

    # 1) 取数（失败直接抛，由 systemd 的重试策略兜底）
    close, volume, pct, factor_panel = build_features(day)
    log.info("取数完成：%d 只标的，%d 个因子", close.shape[1], factor_panel.shape[1])

    # 2) 模型打分
    model = load_model(CFG.model_path)
    scores = score_panel(model, factor_panel)

    # 3) 组合构建：top 20 等权，行业偏离 <= 3%
    weights = build_portfolio(scores, factor_panel["industry"], top_n=CFG.top_n)
    log.info("组合构建完成：%d 只标的，合计权重 %.2f", (weights > 0).sum(), weights.sum())

    # 4) 落库（唯一约束保证幂等）
    n = save_signals(day, weights, scores)
    log.info("信号已入库 %d 条", n)

    # 5) 推送
    push_wecom(day, weights, scores)
    log.info("推送完成")
    return 0

if __name__ == "__main__" 的入口保持极薄，方便用 python -m quant.pipeline --force 手动补跑：

# quant/__main__.py
import argparse
import datetime as dt
import logging
import sys

from quant.pipeline import run_once

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s - %(message)s",
)

parser = argparse.ArgumentParser()
parser.add_argument("--day", type=dt.date.fromisoformat, default=None)
parser.add_argument("--force", action="store_true", help="忽略交易日判断，强制运行")
args = parser.parse_args()
sys.exit(run_once(force=args.force, day=args.day))

四、幂等落库

信号表的唯一约束是幂等的全部保证：

CREATE TABLE IF NOT EXISTS daily_signal (
    trade_date  date             NOT NULL,
    symbol      text             NOT NULL,
    score       double precision NOT NULL,
    weight      double precision NOT NULL,
    model_ver   text             NOT NULL,
    created_at  timestamptz      NOT NULL DEFAULT now(),
    PRIMARY KEY (trade_date, symbol)
);

# quant/db.py
from __future__ import annotations

import datetime as dt

import pandas as pd
from sqlalchemy import create_engine, text

from quant.config import CFG

engine = create_engine(CFG.dsn, pool_pre_ping=True, pool_size=2)

UPSERT = text("""
    INSERT INTO daily_signal (trade_date, symbol, score, weight, model_ver)
    VALUES (:trade_date, :symbol, :score, :weight, :model_ver)
    ON CONFLICT (trade_date, symbol) DO UPDATE
    SET score = EXCLUDED.score,
        weight = EXCLUDED.weight,
        model_ver = EXCLUDED.model_ver
""")


def save_signals(day: dt.date, weights: pd.Series, scores: pd.Series) -> int:
    rows = [
        {"trade_date": day, "symbol": sym, "score": float(scores[sym]),
         "weight": float(w), "model_ver": CFG.model_version}
        for sym, w in weights.items() if w > 0
    ]
    if not rows:
        return 0
    with engine.begin() as conn:
        conn.execute(UPSERT, rows)
    return len(rows)

推送也需要幂等——同一天重复推送会把手机刷屏。用一个简单的 push_log 表打标记：

def already_pushed(day: dt.date, channel: str) -> bool:
    with engine.connect() as conn:
        hit = conn.execute(text(
            "SELECT 1 FROM push_log WHERE trade_date = :d AND channel = :c"
        ), {"d": day, "c": channel}).first()
    return hit is not None

五、推送：企业微信机器人

企业微信群机器人的 webhook 是最省事的通道——不需要申请公众号、不需要服务器回调，一条 HTTP POST 就到手机。

# quant/notify.py
from __future__ import annotations

import datetime as dt
import logging
import time

import requests

from quant.config import CFG
from quant.db import already_pushed, mark_pushed

log = logging.getLogger("quant.notify")


def _fmt_message(day: dt.date, weights, scores) -> str:
    top = weights[weights > 0].sort_values(ascending=False).head(10)
    lines = [f"### 📈 量化信号 {day.isoformat()}",
             f"> 模型版本 `{CFG.model_version}`，持仓 {int((weights > 0).sum())} 只，"
             f"总仓位 {weights.sum():.0%}",
             ""]
    for sym, w in top.items():
        lines.append(f"- `{sym}` 权重 {w:.1%}，得分 {scores[sym]:+.3f}")
    lines.append("")
    lines.append("<font color=\"comment\">本消息由远思码实验室自动生成，不构成投资建议。</font>")
    return "\n".join(lines)


def push_wecom(day: dt.date, weights, scores, retries: int = 3) -> None:
    if CFG.dry_run:
        log.warning("DRY_RUN 开启，跳过推送")
        return
    if already_pushed(day, "wecom"):
        log.info("%s 已推送过，跳过", day)
        return

    payload = {"msgtype": "markdown",
               "markdown": {"content": _fmt_message(day, weights, scores)}}
    last_err: Exception | None = None
    for attempt in range(1, retries + 1):
        try:
            resp = requests.post(CFG.wecom_webhook, json=payload, timeout=5)
            data = resp.json()
            # 企业微信的成功码是 0；频率限制是 45009
            if data.get("errcode") == 0:
                mark_pushed(day, "wecom")
                return
            raise RuntimeError(f"企业微信返回 {data}")
        except Exception as exc:      # noqa: BLE001
            last_err = exc
            wait = 2 ** attempt
            log.warning("推送失败（第 %d 次）：%s，%ds 后重试", attempt, exc, wait)
            time.sleep(wait)
    raise RuntimeError(f"推送最终失败：{last_err}")


def push_alert(msg: str) -> None:
    """告警用：单次尝试、不重试、失败只写日志（告警本身不该再触发告警）。"""
    if CFG.dry_run:
        log.warning("DRY_RUN 开启，告警内容：%s", msg)
        return
    try:
        requests.post(CFG.wecom_webhook,
                      json={"msgtype": "text", "text": {"content": msg}},
                      timeout=5)
    except Exception as exc:      # noqa: BLE001
        log.error("告警推送失败：%s", exc)

webhook 里有 key，不要写进代码库。用 EnvironmentFile 注入，权限 600。

六、systemd：为什么不用 crontab

以前写过 crontab 的用法，但这条流水线更适合 systemd，理由很具体：

需求	cron	systemd timer
依赖别的服务（数据库、网络）	手写 `sleep 30`	`After=network-online.target postgresql.service`
失败重试	自己写包装脚本	`Restart=on-failure` + `RestartSec`
日志	重定向到文件，自己轮转	`journalctl -u` 现成，带结构化输出
立即手动触发	找到那行注释掉再跑	`systemctl start quant-signal.service`
错过补跑（机器重启）	不会补	`Persistent=true`
随机延迟（避免整点打爆 API）	不支持	`RandomizedDelaySec=`
资源限制	无	`MemoryMax` / `CPUQuota`

# /etc/systemd/system/quant-signal.service
[Unit]
Description=Daily quant signal pipeline
Wants=network-online.target
After=network-online.target postgresql.service
# 失败时最多跑 3 次
StartLimitIntervalSec=3600
StartLimitBurst=3

[Service]
Type=oneshot
User=siyuan
Group=siyuan
WorkingDirectory=/home/siyuan/code/python/quant
EnvironmentFile=/etc/quant/quant.env
ExecStart=/home/siyuan/miniconda3/envs/quant/bin/python -m quant
# 超时保护：超过 30 分钟说明卡住了
TimeoutStartSec=1800
Restart=on-failure
RestartSec=120
# 资源兜底，避免把研究机拖死
MemoryMax=6G
CPUQuota=300%
# 最小权限
NoNewPrivileges=true
PrivateTmp=true
ProtectSystem=strict
ProtectHome=read-only
ReadWritePaths=/home/siyuan/code/python/quant/logs
StandardOutput=journal
StandardError=journal
SyslogIdentifier=quant-signal

[Install]
WantedBy=multi-user.target

# /etc/systemd/system/quant-signal.timer
[Unit]
Description=Run quant signal pipeline after market close

[Timer]
# 17:30 触发；A 股 15:00 收盘，盘后数据源一般 16:30 之前更新完毕
OnCalendar=Mon..Fri 17:30:00
# 机器关机期间错过的任务，开机后补跑
Persistent=true
# 随机延迟 0~10 分钟，避免每天同一秒打爆 akshare 接口
RandomizedDelaySec=600
Unit=quant-signal.service

[Install]
WantedBy=timers.target

安装与验证：

sudo systemctl daemon-reload
sudo systemctl enable --now quant-signal.timer

# 看下次触发时间
systemctl list-timers quant-signal.timer

# 立即手动跑一次（验证用）
sudo systemctl start quant-signal.service
journalctl -u quant-signal.service -n 50 --no-pager

# 查看历史所有执行
journalctl -u quant-signal.service --since "7 days ago" | grep -E "完成|失败|跳过"

EnvironmentFile 内容（权限 600，属主 root）：

# /etc/quant/quant.env
QUANT_DSN=postgresql+psycopg2://quant:***@127.0.0.1:5432/quant
WECOM_WEBHOOK=https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=***
MODEL_VERSION=lgb-20261020
DRY_RUN=0

七、心跳与告警：让沉默也能被发现

最危险的故障不是报错，而是什么都没发生。程序悄悄没跑，而你以为它在跑。实验室用一个心跳表解决：

# quant/heartbeat.py
import datetime as dt

from sqlalchemy import text

from quant.db import engine


def beat(name: str, ok: bool, msg: str = "") -> None:
    with engine.begin() as conn:
        conn.execute(text("""
            INSERT INTO heartbeat (task, last_run_at, ok, msg)
            VALUES (:t, now(), :ok, :msg)
            ON CONFLICT (task) DO UPDATE
            SET last_run_at = now(), ok = EXCLUDED.ok, msg = EXCLUDED.msg
        """), {"t": name, "ok": ok, "msg": msg[:500]})

再挂一个每天中午的看门狗 timer：如果 quant-signal 最近一次心跳早于上一个交易日 18:00，就推一条告警。

# quant/watchdog.py
import datetime as dt

from sqlalchemy import text

from quant.calendar import last_trading_day
from quant.db import engine
from quant.notify import push_alert

def check() -> int:
    with engine.connect() as conn:
        row = conn.execute(text(
            "SELECT last_run_at, ok FROM heartbeat WHERE task = 'quant-signal'"
        )).first()
    today = dt.date.today()
    expect_after = dt.datetime.combine(last_trading_day(today), dt.time(18, 0))
    if row is None or row.last_run_at < expect_after:
        push_alert(f"⚠️ 量化信号未在 {expect_after} 前完成，请检查 journalctl -u quant-signal")
        return 1
    if not row.ok:
        push_alert("⚠️ 上次量化信号执行失败，详见日志")
        return 1
    return 0

八、踩过的坑

OnCalendar 的时间是本地时区，但容器里常常是 UTC。如果哪天把服务塞进容器，17:30 会变成凌晨 1:30。
WorkingDirectory 必须显式设置，否则 Type=oneshot 的工作目录是 /，相对路径的配置、模型文件全部找不到。
ProtectHome=read-only 与 conda 环境：conda 在某些操作下要写 ~/.cache，如果策略要写缓存目录，得把 ReadWritePaths 加上。
akshare 接口偶尔 503。重试要带指数退避，而且要区分"接口挂了"和"今天本来没数据"——前者该重试，后者重试只会放大问题。
推送成功 ≠ 你看到了。机器人被移出群、被限流（每分钟 20 条）都会静默失败，所以 push_log 与心跳两者都要有。
别在 ExecStart 里用 shell 管道。ExecStart 不经过 shell，>>、&& 都不会生效；需要 shell 就写包装脚本。

结语

这条链路跑通之后，整个量化研究才算真正闭环：研究 → 回测 → 上线 → 反馈。每天早上八点看手机上那几条消息，比看任何回测曲线都更能暴露问题——真实的价格、真实的公告、真实的滑点，会立刻告诉你模型对不对。

六篇量化系列的导航：

远思为径，码动价值。让它每天自己跑起来，是量化从"研究"变成"系统"的最后一步。