data_cleaner.py

为什么要统一时间格式？因为后续做环比、同比计算时，时间乱了一算就错

import pandas as pd
from datetime import datetime
import re

class DataCleaner:
def __init__(self):
# 这个映射表我找了好久，不同平台订单状态名都不一样
self.status_mapping = {
'WAIT_BUYER_PAY': '待付款',
'WAIT_SELLER_SEND_GOODS': '待发货',
'TRADE_FINISHED': '已完成',
'TRADE_CLOSED': '已关闭',
'AUDITING': '审核中', # 拼多多特有
}

def clean_sales_data(self, df: pd.DataFrame) -> pd.DataFrame:
"""清洗销售数据，返回标准格式DataFrame"""
# 第一步：统一时间格式
# 坑：有的平台时间带毫秒，有的不带
def parse_time(t):
if pd.isna(t):
return None
t = str(t).strip()
# 处理不同格式
formats = [
'%Y-%m-%d %H:%M:%S',
'%Y/%m/%d %H:%M:%S',
'%Y-%m-%dT%H:%M:%S',
'%Y-%m-%d'
]
for fmt in formats:
try:
return datetime.strptime(t, fmt)
except:
continue
return None

df['订单时间'] = df['订单时间'].apply(parse_time)

# 第二步：清洗金额字段
# 有些平台返回"¥99.00"，有些是"99.00元"
def clean_price(price):
if pd.isna(price):
return 0.0
price = str(price)
# 去掉所有非数字和小数点的字符
price = re.sub(r'[^\d.]', '', price)
try:
return float(price)
except:
return 0.0

df['实付金额'] = df['实付金额'].apply(clean_price)

# 第三步：统一订单状态
df['订单状态'] = df['订单状态'].map(self.status_mapping).fillna(df['订单状态'])

# 第四步：去重（防止API重试导致的重复数据）
df = df.drop_duplicates(subset=['订单号'], keep='last')

这个设计真的反人类——不同平台连字段名都不一样。淘宝叫“payment”，拼多多叫“pay_amount”。所以我加了个字段映射配置：

field_mapping.py

核心计算指标

清洗完数据就该算指标了。电商最核心的无非是：GMV、客单价、转化率、复购率、退款率。

有个技巧：别每次都全量计算，用增量更新。我试过每天全量跑500万订单，服务器直接崩了。

metrics_calculator.py

为什么要用rolling window？因为运营要看最近30天的趋势，不是累计值

import pandas as pd
import numpy as np
from datetime import timedelta

class MetricsCalculator:
def __init__(self, df: pd.DataFrame):
self.df = df
# 确保时间索引
self.df['订单时间'] = pd.to_datetime(self.df['订单时间'])
self.df = self.df.set_index('订单时间').sort_index()

def calculate_daily_metrics(self, date: str = None) -> dict:
"""计算每日核心指标"""
if date:
target_date = pd.Timestamp(date)
daily_data = self.df[self.df.index.date == target_date.date()]
else:
daily_data = self.df
target_date = self.df.index.max()

if daily_data.empty:
return {}

# GMV 计算
gmv = daily_data['实付金额'].sum()

# 订单数
order_count = len(daily_data)

# 客单价
avg_order_value = gmv / order_count if order_count > 0 else 0

# 退款率（取最近30天）
thirty_days_ago = target_date - timedelta(days=30)
recent_data = self.df[self.df.index >= thirty_days_ago]
refund_rate = len(recent_data[recent_data['订单状态'] == '已关闭']) / len(recent_data) * 100

# 复购率（取最近90天）
ninety_days_ago = target_date - timedelta(days=90)
buyer_history = self.df[self.df.index >= ninety_days_ago]
buyer_counts = buyer_history.groupby('买家ID')['订单号'].nunique()
repeat_buyers = len(buyer_counts[buyer_counts > 1])
total_buyers = len(buyer_counts)
repeat_rate = repeat_buyers / total_buyers * 100 if total_buyers > 0 else 0

return {
'日期': target_date.strftime('%Y-%m-%d'),
'GMV': round(gmv, 2),
'订单数': order_count,
'客单价': round(avg_order_value, 2),
'退款率': round(refund_rate, 2),
'复购率': round(repeat_rate, 2)
}

def calculate_trend(self, days: int = 30) -> pd.DataFrame:
"""计算趋势数据，用于图表"""
# 用resample做日聚合，比groupby快3倍
daily_metrics = self.df.resample('D').agg({
'实付金额': 'sum',
'订单号': 'count',
'买家ID': pd.Series.nunique
}).rename(columns={
'实付金额': 'GMV',
'订单号': '订单数',
'买家ID': '买家数'
})

# 计算7日移动平均（平滑曲线用）
daily_metrics['GMV_MA7'] = daily_metrics['GMV'].rolling(window=7).mean()

自动化报表生成

这个模块我踩了最深的坑。一开始用openpyxl手动写单元格，代码200多行，改个格式要翻半天。后来发现xlsxwriter + pandas结合，一句代码搞定格式。

（报表样式截图：带图表、条件格式、数据透视表的Excel报表）

report_generator.py

为什么用xlsxwriter？因为支持图表、条件格式、冻结窗格，专业感拉满

import xlsxwriter
import pandas as pd
from io import BytesIO

class ReportGenerator:
def __init__(self):
self.output = BytesIO()
self.workbook = xlsxwriter.Workbook(self.output, {'in_memory': True})

# 定义样式
self.header_format = self.workbook.add_format({
'bold': True,
'bg_color': '#4472C4',
'font_color': 'white',
'border': 1,
'align': 'center',
'valign': 'vcenter',
'font_size': 11
})
self.number_format = self.workbook.add_format({
'num_format': '#,##0.00',
'border': 1,
'align': 'center'
})
self.date_format = self.workbook.add_format({
'num_format': 'yyyy-mm-dd',
'border': 1,
'align': 'center'
})

def generate_daily_report(self, metrics: dict, trend_data: pd.DataFrame) -> BytesIO:
"""生成日报表"""
# Sheet1: 核心指标看板
worksheet1 = self.workbook.add_worksheet('核心指标')
worksheet1.set_column('A:A', 15)
worksheet1.set_column('B:B', 20)
worksheet1.set_column('C:C', 20)
worksheet1.set_column('D:D', 15)

# 写标题
title_format = self.workbook.add_format({
'bold': True,
'font_size': 16,
'align': 'center',
'valign': 'vcenter'
})
worksheet1.merge_range('A1:D1', f"电商日报 - {metrics['日期']}", title_format)

# 写核心指标
headers = ['指标', '数值', '环比', '趋势']
for col, header in enumerate(headers):
worksheet1.write(2, col, header, self.header_format)

# 指标数据
indicators = [
('GMV', metrics['GMV'], '金额'),
('订单数', metrics['订单数'], '数量'),
('客单价', metrics['客单价'], '金额'),
('退款率', metrics['退款率'], '百分比'),
('复购率', metrics['复购率'], '百分比')
]

for row, (name, value, vtype) in enumerate(indicators, start=3):
worksheet1.write(row, 0, name, self.workbook.add_format({'border': 1}))
if vtype == '金额':
worksheet1.write(row, 1, value, self.number_format)
elif vtype == '百分比':
worksheet1.write(row, 1, f"{value:.2f}%",
self.workbook.add_format({'border': 1, 'align': 'center'}))
else:
worksheet1.write(row, 1, int(value),
self.workbook.add_format({'num_format': '#,##0', 'border': 1, 'align': 'center'}))

# Sheet2: 趋势图表
worksheet2 = self.workbook.add_worksheet('趋势分析')

# 写趋势数据
headers = ['日期', 'GMV', '订单数', '买家数', 'GMV_MA7']
for col, header in enumerate(headers):
worksheet2.write(0, col, header, self.header_format)

for row, (date, row_data) in enumerate(trend_data.iterrows(), start=1):
worksheet2.write_datetime(row, 0, date.to_pydatetime(), self.date_format)
worksheet2.write(row, 1, row_data['GMV'], self.number_format)
worksheet2.write(row, 2, row_data['订单数'],
self.workbook.add_format({'num_format': '#,##0', 'border': 1, 'align': 'center'}))
worksheet2.write(row, 3, row_data['买家数'],
self.workbook.add_format({'num_format': '#,##0', 'border': 1, 'align': 'center'}))
worksheet2.write(row, 4, row_data['GMV_MA7'], self.number_format)

# 添加图表
chart = self.workbook.add_chart({'type': 'line'})
chart.add_series({
'name': 'GMV',
'categories': ['趋势分析', 1, 0, len(trend_data), 0],
'values': ['趋势分析', 1, 1, len(trend_data), 1],
})
chart.add_series({
'name': 'GMV_MA7',
'categories': ['趋势分析', 1, 0, len(trend_data), 0],
'values': ['趋势分析', 1, 4, len(trend_data), 4],
})
chart.set_title({'name': 'GMV趋势（含7日移动平均）'})
chart.set_x_axis({'name': '日期'})
chart.set_y_axis({'name': '金额 (元)'})
chart.set_style(10)

worksheet2.insert_chart('G2', chart, {'x_scale': 2, 'y_scale': 1.5})

自动化调度

最后一步，用定时任务每天自动跑。我用的方案是：Python + APScheduler + 企业微信机器人通知。

scheduler.py

为什么用APScheduler？支持cron表达式，比sleep循环稳定100倍

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
import requests

def daily_report_job():
"""每天早上8点自动生成并发送报表"""
try:
# 1. 从平台API拉取数据（这里用模拟数据）
print(f"[{datetime.now()}] 开始拉取数据...")

# 2. 清洗数据
cleaner = DataCleaner()
# 假设 df 是拉到的原始数据
cleaned_df = cleaner.clean_sales_data(df)

# 3. 计算指标
calculator = MetricsCalculator(cleaned_df)
daily_metrics = calculator.calculate_daily_metrics()
trend_data = calculator.calculate_trend(days=30)

# 4. 生成报表
generator = ReportGenerator()
report_bytes = generator.generate_daily_report(daily_metrics, trend_data)

# 5. 保存或发送（这里保存到本地）
with open(f'reports/日报_{daily_metrics["日期"]}.xlsx', 'wb') as f:
f.write(report_bytes.read())

# 6. 企业微信通知
send_wechat_notification(daily_metrics)

print(f"[{datetime.now()}] 日报生成成功！")
except Exception as e:
print(f"日报生成失败: {e}")
# 发告警
send_alert(f"日报生成失败: {str(e)}")

def send_wechat_notification(metrics):
"""发送企业微信机器人消息"""
content = f"""📊 电商日报 - {metrics['日期']}
GMV: ¥{metrics['GMV']:,.2f}
订单数: {metrics['订单数']}
客单价: ¥{metrics['客单价']:.2f}
退款率: {metrics['退款率']:.2f}%
复购率: {metrics['复购率']:.2f}%
"""
# 替换成你的Webhook地址
webhook_url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY"
requests.post(webhook_url, json={"msgtype": "text", "text": {"content": content}})

启动调度

经验总结：你可以立刻用的三个点

这个系统我从搭建到稳定跑了3周，现在每天自动出报表，运营团队直接用Excel透视，再也没人半夜找我要数据了。

（系统运行日志截图：每天早上8点自动生成日报，稳定运行30天）