第三章：Python自动化脚本——运维的瑞士军刀

说实话，在量化运维这个行当里摸爬滚打这么多年，我越来越觉得Python就是运维人员的“第二双手”。你想想看，每天要盯着几十台服务器，检查磁盘、看进程、捞日志……要是全靠手动操作，那不得累死？

我个人习惯是，但凡重复操作超过三次，就写个脚本把它自动化掉。这一章，我就把平时用得最多的几个Python运维场景，掰开揉碎了讲给你听。

3.1 文件与目录操作：运维的基本功

运维工作里，文件操作是最频繁的。日志切割、配置文件备份、数据归档……这些活儿，Python的os和shutil模块都能搞定。

3.1.1 路径拼接与判断

我见过不少新手喜欢用字符串拼接路径，比如"/data/logs/" + filename。嗯，这在Windows上会出大问题。正确的做法是用os.path.join：

import os

# 自动处理不同操作系统的路径分隔符
log_dir = "/data/logs"
today = "2025-03-21"
full_path = os.path.join(log_dir, today, "trade.log")
print(full_path)  # /data/logs/2025-03-21/trade.log

# 判断文件是否存在
if os.path.exists(full_path):
    print("日志文件存在，开始处理...")
else:
    print("文件不存在，请检查日期")

3.1.2 批量重命名与移动

有一次，交易所改了数据文件的命名规则，从trade_20250321.csv变成了20250321_trade.csv。几百个文件，手动改？不存在的。写个脚本几秒钟搞定：

import os
import glob

data_dir = "/data/market_data"
# 匹配所有trade开头的csv文件
for old_name in glob.glob(os.path.join(data_dir, "trade_*.csv")):
    # 提取日期部分
    basename = os.path.basename(old_name)
    date_part = basename.split("_")[1].split(".")[0]
    new_name = os.path.join(data_dir, f"{date_part}_trade.csv")
    os.rename(old_name, new_name)
    print(f"已重命名: {basename} -> {os.path.basename(new_name)}")

3.1.3 目录遍历与文件筛选

量化系统跑久了，日志目录里会堆积大量文件。我们需要定期清理超过30天的旧日志。用os.walk可以递归遍历所有子目录：

import os
import time

def clean_old_logs(log_root, days=30):
    now = time.time()
    cutoff = now - days * 86400
    cleaned_count = 0
    
    for dirpath, dirnames, filenames in os.walk(log_root):
        for filename in filenames:
            if filename.endswith(".log"):
                full_path = os.path.join(dirpath, filename)
                # 获取文件最后修改时间
                mtime = os.path.getmtime(full_path)
                if mtime < cutoff:
                    os.remove(full_path)
                    cleaned_count += 1
                    print(f"已删除过期日志: {full_path}")
    
    return cleaned_count

# 执行清理
count = clean_old_logs("/data/logs", days=30)
print(f"本次共清理 {count} 个日志文件")

3.2 系统信息采集：给服务器做“体检”

量化交易对系统性能极其敏感。CPU飙高、内存不足、磁盘写满……任何一个指标出问题，都可能导致交易延迟甚至亏损。所以，系统信息采集是运维自动化的核心环节。

3.2.1 使用psutil采集系统指标

psutil是我最常用的系统监控库，没有之一。它跨平台、接口统一，用起来非常顺手：

import psutil
import datetime

def collect_system_info():
    info = {}
    
    # CPU信息
    info['cpu_percent'] = psutil.cpu_percent(interval=1)
    info['cpu_count'] = psutil.cpu_count()
    info['cpu_load_avg'] = psutil.getloadavg()
    
    # 内存信息
    mem = psutil.virtual_memory()
    info['memory_total'] = mem.total / (1024**3)  # 转GB
    info['memory_used'] = mem.used / (1024**3)
    info['memory_percent'] = mem.percent
    
    # 磁盘信息
    disk = psutil.disk_usage('/')
    info['disk_total'] = disk.total / (1024**3)
    info['disk_used'] = disk.used / (1024**3)
    info['disk_percent'] = disk.percent
    
    # 网络信息
    net = psutil.net_io_counters()
    info['net_bytes_sent'] = net.bytes_sent
    info['net_bytes_recv'] = net.bytes_recv
    
    # 时间戳
    info['timestamp'] = datetime.datetime.now().isoformat()
    
    return info

# 采集一次系统信息
sys_info = collect_system_info()
for key, value in sys_info.items():
    print(f"{key}: {value}")

3.2.2 进程管理：监控交易程序

量化交易程序必须7x24小时运行。如果进程挂了，我们需要第一时间发现并重启。用psutil可以轻松实现进程监控：

import psutil

def check_trading_process(process_name="quant_trader"):
    """检查交易进程是否在运行"""
    for proc in psutil.process_iter(['pid', 'name', 'create_time']):
        try:
            if process_name in proc.info['name']:
                return {
                    'status': 'running',
                    'pid': proc.info['pid'],
                    'uptime': datetime.datetime.now() - datetime.datetime.fromtimestamp(
                        proc.info['create_time']
                    )
                }
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            pass
    return {'status': 'stopped'}

# 检查交易程序
result = check_trading_process()
if result['status'] == 'running':
    print(f"交易程序运行正常，PID: {result['pid']}，已运行: {result['uptime']}")
else:
    print("警告：交易程序未运行！")
    # 这里可以触发重启逻辑

3.3 异常处理与日志：让脚本更健壮

运维脚本最怕什么？怕半夜跑着跑着突然挂了，连个错误信息都没留下。所以，异常处理和日志记录是脚本的“安全带”和“黑匣子”。

3.3.1 异常处理：别让脚本静默死亡

很多新手写脚本，遇到异常就直接崩溃。其实，合理的异常处理可以让脚本在出错时优雅地降级：

import os
import sys

def safe_read_config(config_path):
    """安全读取配置文件"""
    try:
        with open(config_path, 'r') as f:
            return f.read()
    except FileNotFoundError:
        print(f"配置文件 {config_path} 不存在，使用默认配置")
        return "default_config"
    except PermissionError:
        print(f"没有权限读取 {config_path}，请检查文件权限")
        sys.exit(1)
    except Exception as e:
        print(f"读取配置文件时发生未知错误: {e}")
        # 记录到日志文件
        with open('/var/log/quant_ops.log', 'a') as log:
            log.write(f"[ERROR] {datetime.datetime.now()}: {e}\n")
        return None

3.3.2 日志记录：用logging模块代替print

我早期写脚本也喜欢用print，但后来发现这玩意儿在生产环境根本不够用。日志需要分级、需要输出到文件、需要自动轮转……logging模块才是正道：

import logging
import logging.handlers

def setup_logger(log_file="/var/log/quant_ops.log"):
    """配置日志系统"""
    logger = logging.getLogger('quant_ops')
    logger.setLevel(logging.DEBUG)
    
    # 文件处理器：每天轮转，保留30天
    file_handler = logging.handlers.TimedRotatingFileHandler(
        log_file, when='midnight', backupCount=30
    )
    file_handler.setLevel(logging.INFO)
    
    # 控制台处理器：开发时用
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.DEBUG)
    
    # 日志格式
    formatter = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    file_handler.setFormatter(formatter)
    console_handler.setFormatter(formatter)
    
    logger.addHandler(file_handler)
    logger.addHandler(console_handler)
    
    return logger

# 使用日志
logger = setup_logger()
logger.info("系统信息采集任务开始")
try:
    info = collect_system_info()
    logger.info(f"采集完成，CPU使用率: {info['cpu_percent']}%")
except Exception as e:
    logger.error(f"采集失败: {e}", exc_info=True)

3.4 知识体系总览

说了这么多，我们来梳理一下本章的核心逻辑。下面这张图，是我自己总结的Python运维自动化知识框架：

说白了，这三个模块就是运维自动化的三根支柱。文件操作管的是“数据怎么存”，系统采集管的是“机器跑得怎么样”，异常日志管的是“出错了怎么办”。三者缺一不可。

我个人习惯是，每写一个运维脚本，先把日志系统搭好，再把关键操作包在try/except里。这样即使脚本半夜跑崩了，第二天早上我也能通过日志快速定位问题。嗯，这个习惯帮我避免了好几次生产事故。

---