7.13 自动化流水线与成本控制

手动运行只能算实验，自动化才是系统。这一篇讲怎么让它自己跑起来，并且省钱。

一、四条自动化流水线#

二、调度实现（三种方案）#

方案一：系统 Cron（最稳）#

# 每天早上 8:05 运行
5 8 * * * cd /path/to/project && /usr/bin/python3 run_daily.py >> logs/daily.log 2>&1

# 每天晚上 23:30 运行复盘
30 23 * * * cd /path/to/project && /usr/bin/python3 run_review.py >> logs/review.log 2>&1

方案二：APScheduler（纯 Python，跨平台）#

from apscheduler.schedulers.blocking import BlockingScheduler
import logging

logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s %(levelname)s %(message)s')

sched = BlockingScheduler(timezone='Asia/Shanghai')

@sched.scheduled_job('cron', hour=8, minute=5)
def job_daily():
    logging.info('开始每日采集')
    try:
        run_daily()
    except Exception as e:
        logging.exception(f'每日任务失败: {e}')

@sched.scheduled_job('cron', hour=23, minute=30)
def job_review():
    logging.info('开始赛后复盘')
    try:
        run_review()
    except Exception as e:
        logging.exception(f'复盘任务失败: {e}')

if __name__ == '__main__':
    logging.info('调度器启动')
    sched.start()

方案三：单实例启动器（项目规范推荐）#

如果项目需要长期后台运行且要防止重复启动，参考项目内的单实例启动器规范：PID 文件 + 端口检测 + 进程存活三重验证。

三、幂等与容错#

幂等设计#

def run_step(name, fn, conn):
    """通用步骤执行器：带幂等检查与日志"""

    # 已执行过就跳过
    if is_done(name, conn):
        logging.info(f'[{name}] 已完成，跳过')
        return True

    try:
        logging.info(f'[{name}] 开始')
        fn()
        mark_done(name, conn)
        logging.info(f'[{name}] 完成')
        return True
    except Exception as e:
        logging.exception(f'[{name}] 失败: {e}')
        send_alert(f'步骤 {name} 失败: {e}')   # 失败要告警，不能静默
        return False

三个必须的日志项#

四、成本控制（核心章节）#

四层降本策略#

缓存策略设计#

# 不同数据的缓存时长应当不同
CACHE_TTL = {
    'fixtures_history': 30 * 86400,   # 历史赛程：不会变，缓存 30 天
    'match_stats':      30 * 86400,   # 已结束比赛统计：不会变，30 天
    'match_pack':       30 * 86400,   # 已结束比赛全景：30 天
    'team_advanced':     3 * 86400,   # 球队高阶画像：3 天
    'injuries':            2 * 3600,  # 伤停：2 小时（会变）
    'today_digest':          1800,    # 今日总览：30 分钟
    'quota':                    0,    # 余额：不缓存
}

def ttl_for(kind):
    return CACHE_TTL.get(kind, 3600)

关键原则：区分「不变数据」和「易变数据」。
已结束比赛的数据永远不变，可以长期缓存。
伤停、数据数值、今日赛程是易变的，缓存时间要短。
把两者用同一个 TTL 处理，要么浪费点数，要么用到过期数据。

成本预算估算#

PRICE = {                       # 各接口点数（与接口文档一致）
    'today_digest':   5,
    'match_pack':    50,
    'team_advanced': 10,
    'team_injuries':  5,
    'team_squad':     3,
    'fixtures':       2,
}

def estimate_daily(n_matches=3):
    """估算一天的消耗"""
    cost = PRICE['today_digest']                       # 1 次总览
    for _ in range(n_matches):
        cost += PRICE['match_pack']                    # 每场全景
        cost += PRICE['team_injuries'] * 2             # 双方伤停
    return cost

print(f'每日预估消耗: {estimate_daily()} 点')

五、成本告警#

def check_quota_and_alert(threshold=500):
    """余额低于阈值时告警"""
    q = fetch('/quota', cache_key=None)
    remain = q.get('credits', q.get('balance', 0))
    if remain < threshold:
        send_alert(f'点数余额仅剩 {remain}，请及时获取服务')
    return remain

建议：把余额检查放在每次任务的开始，避免跑到一半因余额不足而失败。

六、合规检查自动化#

把合规检查也做成流水线的一环：

FORBIDDEN_PATTERNS = [
    '必胜', '稳赢', '包中', '核心首选', '一定赢',
    '内幕', '必中', '稳赚', '跟上',
]

def compliance_check(text):
    """基础合规检查：命中禁用词就拦截"""
    hits = [w for w in FORBIDDEN_PATTERNS if w in text]
    return {'passed': not hits, 'hits': hits}

注意：关键词过滤只是第一道防线。
它拦不住改写后的违规表述（比如把「必赢」改成「赢面极大」）。
正式发布前，仍建议用 AI 做一次语义层面的合规自检（见 6.6）。

七、自动化成熟度分级#

大多数个人项目做到 L3 就足够了。不要过度工程化。

下一步该看什么#

• 验证你的策略是否有效 → 7.14 回测验证
• 把系统变成产品 → 8.2 常见问题