7.12 30 行代码：搭建每日赛事预警机器人

这一篇给一个完整可运行的最小系统。你可以直接复制运行，理解后再改造。

一、目标#

每天早上自动做三件事：

1. 拉取当天赛事；
2. 筛选出「市场存在分歧」的场次；
3. 输出一份可直接阅读的清单。

二、核心代码（30 行）#

import os, requests, json
from pathlib import Path

API_KEY = os.environ.get('QIUXIAOCE_API_KEY', 'YOUR_API_KEY_HERE')
BASE = 'https://www.qiuxiaoce.com/wp-json/abv2-creator/v1'
HEADERS = {'X-API-Key': API_KEY, 'Accept': 'application/json'}
CACHE = Path('./cache'); CACHE.mkdir(exist_ok=True)


def fetch(path, cache_key=None, params=None):
    """带缓存请求：历史数据不重复扣点"""
    if cache_key and (CACHE / f'{cache_key}.json').exists():
        return json.loads((CACHE / f'{cache_key}.json').read_text(encoding='utf-8'))
    r = requests.get(f'{BASE}{path}', headers=HEADERS, params=params, timeout=20)
    r.raise_for_status()
    data = r.json()
    if cache_key:
        (CACHE / f'{cache_key}.json').write_text(
            json.dumps(data, ensure_ascii=False), encoding='utf-8')
    return data


def impl(odds):
    """市场参考值转隐含概率"""
    return 1.0 / odds if odds and odds > 1 else None


def scan_alerts(date):
    """扫描当日赛事，输出值得关注的场次"""
    data = fetch('/today/digest', cache_key=f'digest_{date}')
    alerts = []
    for m in data.get('fixtures', data.get('data', [])):
        h, d, a = impl(m.get('home_odds')), impl(m.get('draw_odds')), impl(m.get('away_odds'))
        if not all([h, d, a]):
            continue
        total = h + d + a
        overround = total - 1                      # 市场摩擦成本率
        fair = sorted([h / total, d / total, a / total], reverse=True)
        # 关注条件：市场摩擦成本异常 或 前三项概率接近（分歧大）
        if overround > 0.10 or (fair[0] - fair[-1]) < 0.10:
            alerts.append({
                'match': f"{m.get('home_team')} vs {m.get('away_team')}",
                'league': m.get('league'),
                'time': m.get('kickoff'),
                'overround': round(overround, 4),
                'spread': round(fair[0] - fair[-1], 4),
            })
    return alerts


if __name__ == '__main__':
    for a in scan_alerts('2026-09-13'):
        print(f"[{a['time']}] {a['league']} | {a['match']}")
        print(f"    市场摩擦成本 {a['overround']:.2%} | 分歧度 {a['spread']:.2%}")

三、代码解读#

四、筛选条件的原理#

条件一：市场摩擦成本率异常（> 10%）#

正常市场摩擦成本在 3%~8%。如果某场超过 10%，说明定价可能存在问题，或者数据源有异常。

条件二：三项概率接近（最大与最小差 < 10%）#

这意味着市场认为三种结果的可能性接近，这类比赛往往是最难判断、也最容易出现分歧的场次。

五、四个改造方向#

方向一：接入更多筛选维度#

# 加入伤停维度：缺阵核心球员的场次也值得关注
def has_key_injury(team_id):
    inj = fetch(f'/teams/{team_id}/injuries', cache_key=f'inj_{team_id}')
    return len(inj.get('injuries', inj.get('data', []))) > 0

方向二：输出 HTML 报告#

def to_html(alerts):
    rows = ''.join(
        f"{a['time']}{a['league']}"
        f"{a['match']}{a['overround']:.2%}"
        for a in alerts
    )
    return f'{rows}
时间
联赛
比赛
市场摩擦成本
'

方向三：接入推送#

方向四：交给 AI 生成完整报告#

def gen_report(alerts):
    """把预警清单交给 AI，生成可读报告（需自备模型 API）"""
    prompt = (
        '以下是今日筛选出的赛事清单，请生成一份简报：n'
        + json.dumps(alerts, ensure_ascii=False, indent=2)
        + 'nn要求：客观陈述，不预测比分，不给具体操作建议。'
    )
    # 调用你自己的模型接口
    return call_llm(prompt)

六、三个工程注意点#

注意点一：缓存键要包含日期#

cache_key=f'digest_{date}'——否则第二天的数据会命中前一天的缓存。

注意点二：先查余额#

def check_quota():
    q = fetch('/quota', cache_key=None)   # 0 点，不要缓存
    return q

注意点三：异常不中断#

单场比赛数据缺失不应该导致整个脚本失败。用 try/except 包住每场比赛的处理。

七、下一步可以做什么#

下一步该看什么#

• 让它每天自动跑 → 7.13 自动化流水线与成本控制
• 验证你的筛选是否真的有效 → 7.14 回测验证