第 43 篇：策略监控日志告警与异常处理

# ---------- 1. logging 四件套：一次配置，全程复用 ----------
import logging
from pathlib import Path

def setup_logger(name="quant", log_file="logs/run.log"):
    Path("logs").mkdir(exist_ok=True)
    logger = logging.getLogger(name)
    logger.setLevel(logging.DEBUG)            # 入口放开到DEBUG
    if logger.handlers:                       # 防重复添加handler
        return logger
    fmt = logging.Formatter(
        "%(asctime)s [%(levelname)s] %(name)s - %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S")
    fh = logging.FileHandler(log_file, encoding="utf-8")
    fh.setLevel(logging.DEBUG)                # 文件记全量
    sh = logging.StreamHandler()
    sh.setLevel(logging.INFO)                 # 屏幕只看INFO以上
    fh.setFormatter(fmt); sh.setFormatter(fmt)
    logger.addHandler(fh); logger.addHandler(sh)
    return logger

log = setup_logger()

# ---------- 2. 异常分类处理：重试 + 降级 ----------
import time
import numpy as np
import pandas as pd

class TransientError(Exception):  # 瞬时性：值得重试
    pass

def fetch_with_retry(fetch_fn, retries=3, backoff=2.0, tag="data"):
    """带指数退避的重试：第n次等 backoff*n 秒"""
    for attempt in range(1, retries + 1):
        try:
            return fetch_fn()
        except TransientError as e:
            if attempt == retries:
                log.error(f"[{tag}] 重试{retries}次仍失败: {e}")
                raise
            wait = backoff * attempt
            log.warning(f"[{tag}] 第{attempt}次失败, {wait:.0f}s后重试: {e}")
            time.sleep(wait)

def get_price_with_monitoring(code):
    """拉行情：瞬时错误重试，环境错误降级到本地缓存"""
    try:
        # 模拟一个三成概率超时的接口（教学用，真实场景替换为akshare）
        if np.random.default_rng().random() < 0.3:
            raise TransientError("接口超时(模拟)")
        return pd.Series({"600519": 1520.0})  # 模拟返回
    except TransientError:
        return fetch_with_retry(
            lambda: pd.Series({"600519": 1520.0}),
            tag=f"price:{code}")
    except Exception as e:                    # 环境性：降级
        log.warning(f"[price:{code}] 接口异常，降级读本地缓存: {e}")
        cache = Path("data/raw") / f"{code}.parquet"
        if cache.exists():
            df = pd.read_parquet(cache)
            return df["close"].iloc[-1:]
        log.error(f"[price:{code}] 本地缓存也不存在！")
        return None

# ---------- 3. 心跳与看门狗 ----------
from datetime import datetime, timedelta

HEARTBEAT_FILE = Path("logs/heartbeat.txt")

def write_heartbeat(task_name="daily_update"):
    """任务正常结束时调用：报平安"""
    HEARTBEAT_FILE.parent.mkdir(exist_ok=True)
    ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    HEARTBEAT_FILE.write_text(f"{ts}|{task_name}", encoding="utf-8")
    log.info(f"心跳已写入: {ts} ({task_name})")

def check_heartbeat(max_silence_hours=26):
    """看门狗：超时未报平安 → 告警"""
    if not HEARTBEAT_FILE.exists():
        return "CRITICAL: 心跳文件不存在——任务从未成功运行过"
    ts_str, task = HEARTBEAT_FILE.read_text(
        encoding="utf-8").strip().split("|")
    last = datetime.strptime(ts_str, "%Y-%m-%d %H:%M:%S")
    silent = datetime.now() - last
    if silent > timedelta(hours=max_silence_hours):
        return f"CRITICAL: {task} 已静默 {silent}, 任务疑似挂死"
    return f"OK: {task} 上次心跳 {ts_str}"

def alert(message):
    """教学告警：写文件 + 弹控制台响铃。真实环境换webhook推送"""
    with open("logs/alerts.log", "a", encoding="utf-8") as f:
        f.write(f"{datetime.now():%Y-%m-%d %H:%M:%S} {message}\n")
    print(f"\a>>> 告警: {message}")

# ---------- 4. 每日运行状态表 ----------
def run_daily_job():
    """每日任务的完整监控骨架"""
    status = {"date": datetime.now().strftime("%Y-%m-%d"),
              "data": "-", "signal": "-", "error": ""}
    try:
        px = get_price_with_monitoring("600519")
        if px is None:
            status["data"] = "FAIL"
            alert("行情获取失败且无缓存")
        else:
            status["data"] = "OK"
            log.info(f"行情获取成功: {px.iloc[0]}")
        # ... 策略信号计算（略，见第50篇整合）...
        status["signal"] = "OK"
        write_heartbeat("daily_job")           # 全程无异常 → 报平安
    except Exception as e:
        status["error"] = str(e)[:80]
        status["signal"] = "FAIL"
        alert(f"每日任务异常终止: {e}")
    # 状态追加到CSV，积累成健康度档案
    row = pd.DataFrame([status])
    status_file = Path("logs/daily_status.csv")
    row.to_csv(status_file, mode="a", header=not status_file.exists(),
               index=False, encoding="utf-8")
    return status

status = run_daily_job()
print(status)
print(check_heartbeat())

示例输出（示意）:
2026-09-28 09:30:00 [INFO] quant - 行情获取成功: 1520.0
2026-09-28 09:30:00 [INFO] quant - 心跳已写入: 2026-09-28 09:30:00 (daily_job)
{'date': '2026-09-28', 'data': 'OK', 'signal': 'OK', 'error': ''}
OK: daily_job 上次心跳 2026-09-28 09:30:00
（当接口连续超时时，会看到：）
2026-09-28 09:30:04 [WARNING] quant - [price:600519] 第1次失败, 2s后重试: 接口超时(模拟)
2026-09-28 09:30:08 [WARNING] quant - [price:600519] 第2次失败, 4s后重试: 接口超时(模拟)