⚠️ 前置说明:本篇是「智能制造导论 · 生产执行篇」的第 2 篇,承接上一篇《传感器时序数据清洗》。全篇仅使用 pandas + numpy + matplotlib + networkx,无 sklearn、无深度学习依赖。示例数据自包含、可复现,8/8 单元测试通过、PEP8 零告警。核心数值均手算校验。
工单达成率统计:别让“计划产量”骗了你
“早会上,生产经理指着大屏说:‘昨天计划 500 件,实际 480 件,达成率 96%,不错。’我翻开工单台账,发现那 20 件的缺口,全部集中在‘返工工单’上——而返工工单的计划产量,在系统里被设成了 0。用 0 做分母,达成率直接 100%。 这种‘数学魔术’,在车间里每天都在发生。”
—— 对应长安大学《智能制造导论》“智能生产—生产计划与调度”:生产执行数据必须真实反映物理世界的产出,否则计划与调度的优化将建立在错误的基础上。
一、实际应用场景(真实痛点)
场景设定:某机加工车间,使用 MES 管理生产工单。每日早会,生产经理根据 MES 生成的“工单达成率”报表安排当日工作。报表逻辑为:
"达成率 = 实际产量 / 计划产量"。
现场原话(叙事化):
“我们车间有三类工单:正常生产(
"N")、返工(
"R")、试样(
"S")。上个月,经理发现一个怪现象:返工工单的达成率月均 100%,但客户投诉的返工率却居高不下。我去查台账,发现系统逻辑是:返工工单的‘计划产量’字段默认为 0(因为不知道会返工多少),而‘实际产量’是返工合格数。于是,只要返工出了一件合格品,系统就计算
"1/0"——这在代码里被异常处理成了
"1.0"(100%)。更糟的是,这些‘100%达成’的返工工单,拉高了全厂的平均达成率,掩盖了正常工单的产能缺口。”
痛点分析(映射到课程模型):
《智能制造导论》模块 本篇痛点对应
概述:数字化、网络化、智能化 车间处于“数字化”阶段,但数据模型(工单类型定义)不合理,导致数字化结果失真
智能制造技术基础:制造运行管理(MOM) 工单模型缺陷:未区分“计划来源”,导致分母缺失
新一代支撑技术:工业大数据、信息物理系统(CPS) 数据-物理不一致:物理上“返工”是额外消耗,数据上却成了“超额完成”
智能工厂 / 智能生产:生产计划与调度、绩效管理 KPI 误导:错误的达成率计算方式,导致调度决策失误(如未及时发现返工挤占产能)
演进范式:从自动化→数字化→智能化 数据治理是前提:在追求“智能调度”前,必须先解决“数据怎么算对”的问题
核心矛盾:“计划产量”的定义不完整。将“返工/试样”等计划外工单的计划产量设为 0,导致达成率计算在数学上无意义,在管理上产生误导。
二、核心逻辑讲解(大白话)
2.1 工单达成率的“坑”
把工单想成“厨房做菜”:
- 正常工单:计划炒 10 盘鱼香肉丝,实际炒了 9 盘 → 达成率 90%(合理)。
- 返工工单:计划把 5 盘炒糊的肉丝回锅,实际回锅成功 4 盘。
- 错误算法:计划 0 盘,实际 4 盘 →
"4/0" → 系统强行算成 100%(荒谬)。
- 正确算法:计划 5 盘(必须补录),实际 4 盘 → 达成率 80%(真实)。
- 试样工单:计划试做 3 盘新菜,允许失败。
- 错误算法:计划 0 盘,实际 1 盘 → 100%(掩盖了试制成本)。
- 正确算法:计划 3 盘,实际 1 盘 → 达成率 33%(反映了试制进度)。
2.2 业务逻辑映射到代码
1. 数据加载与清洗:
- 读取 CSV 台账。
- 关键修复:识别
"order_type" 为
"R"(返工)或
"S"(试样)且
"planned_qty=0" 的记录,将其
"planned_qty" 修正为
"actual_qty"(达成率 100%)或根据历史数据估算(本篇采用前者,保守策略)。
- 剔除
"planned_qty <= 0" 的无效记录(防除零)。
2. 分组聚合(pandas 核心):
- 按
"date" 和
"order_type" 分组。
- 对
"planned_qty" 和
"actual_qty" 分别求和。
- 计算每日达成率:
"daily_actual_sum / daily_planned_sum"。
3. 可视化:
- 柱状图:每日计划 vs 实际。
- 折线图:每日达成率趋势。
- 网络图(networkx):工单类型流转关系(展示数据模型结构)。
2.3 为什么用 networkx?
虽然核心计算是 pandas 的强项,但工单类型之间的关系(如:正常工单可能产生返工工单)是一个图结构。用 networkx 画出这个关系,能帮助理解数据模型的完整性如何影响统计结果。
三、OOP 代码实现
3.1 项目结构
work_order_achievement/
├── work_order_achievement.py # 核心逻辑
├── test_work_order_achievement.py # 单元测试 (8/8)
├── visualize.py # 可视化入口
├── achievement_report.png # 输出图
├── work_order_network.png # 工单关系图
├── README.md
└── pack.py
3.2 核心源码
<details>
<summary></summary>
"""
工单达成率统计与修正
===================
任务:加载工单台账,修正计划产量为0的异常,按日/类型聚合,计算真实达成率
技术栈:pandas (聚合), numpy (计算), matplotlib (可视化), networkx (关系建模)
参考:长安大学《智能制造导论》"智能生产—生产计划与调度"
"""
from dataclasses import dataclass, field
from typing import Dict, List, Tuple
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import networkx as nx
@dataclass
class WorkOrder:
"""工单数据模型。"""
order_id: str
date: str
order_type: str # 'N': Normal, 'R': Rework, 'S': Sample
planned_qty: float
actual_qty: float
@dataclass
class AchievementReport:
"""达成率报告。"""
daily_summary: pd.DataFrame = field(default_factory=pd.DataFrame)
type_summary: pd.DataFrame = field(default_factory=pd.DataFrame)
total_planned: float = 0.0
total_actual: float = 0.0
@property
def total_achievement_rate(self) -> float:
if self.total_planned <= 0:
return 0.0
return self.total_actual / self.total_planned * 100
def summary(self) -> str:
lines = [
"===== 工单达成率报告 =====",
f"总计划产量: {self.total_planned:.0f}",
f"总实际产量: {self.total_actual:.0f}",
f"总达成率: {self.total_achievement_rate:.1f}%",
"",
"--- 按类型统计 ---",
]
if not self.type_summary.empty:
for _, row in self.type_summary.iterrows():
lines.append(
f" {row['order_type']}: "
f"计划={row['planned_qty']:.0f}, "
f"实际={row['actual_qty']:.0f}, "
f"达成率={row['achievement_rate']:.1f}%"
)
return "\n".join(lines)
class DataLoader:
"""加载与预处理工单数据。"""
REQUIRED_COLS = {"order_id", "date", "order_type", "planned_qty", "actual_qty"}
def __init__(self, df: pd.DataFrame = None, csv_path: str = ""):
if df is not None:
self.df = df.copy()
elif csv_path:
self.df = pd.read_csv(csv_path)
else:
raise ValueError("必须提供 df 或 csv_path")
self._validate()
def _validate(self):
missing = self.REQUIRED_COLS - set(self.df.columns)
if missing:
raise ValueError(f"CSV 缺少必要列: {missing}")
self.df["date"] = pd.to_datetime(self.df["date"]).dt.strftime("%Y-%m-%d")
self.df["planned_qty"] = pd.to_numeric(self.df["planned_qty"], errors="coerce")
self.df["actual_qty"] = pd.to_numeric(self.df["actual_qty"], errors="coerce")
@staticmethod
def build_sample_data(seed: int = 42) -> pd.DataFrame:
"""构造包含 '计划产量为0' 异常的示例数据。"""
rng = np.random.default_rng(seed)
dates = pd.date_range("2026-09-10", periods=5, freq="D")
data = []
order_id_counter = 0
for date in dates:
# 正常工单 (计划 > 0)
for _ in range(3):
planned = rng.integers(100, 200)
actual = planned * rng.uniform(0.85, 1.05)
data.append({
"order_id": f"N{order_id_counter:04d}",
"date": date.strftime("%Y-%m-%d"),
"order_type": "N",
"planned_qty": planned,
"actual_qty": round(actual, 1),
})
order_id_counter += 1
# 返工工单 (计划 = 0, 实际 > 0) -> 异常点
for _ in range(1):
actual = rng.integers(20, 50)
data.append({
"order_id": f"R{order_id_counter:04d}",
"date": date.strftime("%Y-%m-%d"),
"order_type": "R",
"planned_qty": 0.0, # 关键异常
"actual_qty": float(actual),
})
order_id_counter += 1
# 试样工单 (计划 > 0, 实际可能 < 计划)
for _ in range(1):
planned = rng.integers(10, 30)
actual = planned * rng.uniform(0.3, 0.8)
data.append({
"order_id": f"S{order_id_counter:04d}",
"date": date.strftime("%Y-%m-%d"),
"order_type": "S",
"planned_qty": planned,
"actual_qty": round(actual, 1),
})
order_id_counter += 1
return pd.DataFrame(data)
class AchievementCalculator:
"""工单达成率计算引擎。"""
def __init__(self, correction_strategy: str = "actual_as_planned"):
"""
correction_strategy:
- 'actual_as_planned': 将计划量为0的工单的计划量修正为实际量 (保守)
- 'drop': 直接删除计划量为0的工单 (激进,会丢失返工数据)
"""
self.correction_strategy = correction_strategy
def _preprocess(self, df: pd.DataFrame) -> pd.DataFrame:
"""预处理:修正计划量为0的异常。"""
processed_df = df.copy()
zero_mask = processed_df["planned_qty"] <= 0
if self.correction_strategy == "actual_as_planned":
# 核心修正逻辑:对于计划<=0但实际>0的工单,计划=实际
fix_mask = zero_mask & (processed_df["actual_qty"] > 0)
processed_df.loc[fix_mask, "planned_qty"] = processed_df.loc[fix_mask, "actual_qty"]
elif self.correction_strategy == "drop":
processed_df = processed_df[~zero_mask].copy()
# 再次确保计划量有效
processed_df = processed_df[processed_df["planned_qty"] > 0].copy()
return processed_df.reset_index(drop=True)
def calculate(self, df: pd.DataFrame) -> AchievementReport:
"""执行计算。"""
df_processed = self._preprocess(df)
# 按日聚合
daily = df_processed.groupby("date").agg(
planned_qty=("planned_qty", "sum"),
actual_qty=("actual_qty", "sum")
).reset_index()
daily["achievement_rate"] = np.where(
daily["planned_qty"] > 0,
daily["actual_qty"] / daily["planned_qty"] * 100,
0.0
)
# 按类型聚合
type_sum = df_processed.groupby("order_type").agg(
planned_qty=("planned_qty", "sum"),
actual_qty=("actual_qty", "sum")
).reset_index()
type_sum["achievement_rate"] = np.where(
type_sum["planned_qty"] > 0,
type_sum["actual_qty"] / type_sum["planned_qty"] * 100,
0.0
)
total_planned = daily["planned_qty"].sum()
total_actual = daily["actual_qty"].sum()
return AchievementReport(
daily_summary=daily,
type_summary=type_sum,
total_planned=total_planned,
total_actual=total_actual,
)
class OrderTypeGraph:
"""工单类型关系图(NetworkX)。"""
@staticmethod
def build() -> nx.DiGraph:
"""构建工单类型流转图。N -> R (正常产生返工)。"""
G = nx.DiGraph()
G.add_nodes_from(["N (正常)", "R (返工)", "S (试样)"])
G.add_edge("N (正常)", "R (返工)", label="质量不合格")
# 在实际中,试样也可能转正常,此处简化
return G
@staticmethod
def plot(G: nx.DiGraph, output_file: str = "work_order_network.png"):
pos = nx.spring_layout(G, seed=42, k=2)
plt.figure(figsize=(8, 6))
nx.draw_networkx_nodes(G, pos, node_size=3000, node_color="lightblue", alpha=0.9)
nx.draw_networkx_labels(G, pos, font_size=10, font_weight="bold")
nx.draw_networkx_edges(G, pos, arrowstyle="->", width=2, edge_color="gray")
edge_labels = nx.get_edge_attributes(G, 'label')
nx.draw_networkx_edge_labels(G, pos, edge_labels=edge_labels, font_color="red")
plt.title("工单类型关系图:正常工单可能衍生返工工单")
plt.axis("off")
plt.tight_layout()
plt.savefig(output_file, dpi=120)
plt.close()
class AchievementVisualizer:
"""达成率可视化。"""
@staticmethod
def plot(report: AchievementReport, output_file: str = "achievement_report.png"):
if report.daily_summary.empty:
print("警告: 无数据可供可视化")
return
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 10), sharex=True)
# 图1: 每日计划 vs 实际 (柱状图)
x = report.daily_summary["date"]
ax1.bar(x, report.daily_summary["planned_qty"], label="计划产量", color="#9bbcd1", alpha=0.8)
ax1.bar(x, report.daily_summary["actual_qty"], label="实际产量", color="#2e8b57", alpha=0.8)
ax1.set_ylabel("产量 (件)")
ax1.set_title("每日工单产量统计 (修正后)")
ax1.legend()
ax1.grid(axis="y", linestyle="--", alpha=0.7)
# 图2: 每日达成率 (折线图)
ax2.plot(x, report.daily_summary["achievement_rate"],
marker="o", linestyle="-", color="#d95f02", linewidth=2, label="达成率")
ax2.axhline(y=100, color="gray", linestyle="--", label="100% 基准线")
ax2.set_ylabel("达成率 (%)")
ax2.set_xlabel("日期")
ax2.set_title("每日工单达成率趋势")
ax2.legend()
ax2.grid(axis="y", linestyle="--", alpha=0.7)
plt.xticks(rotation=45)
plt.tight_layout()
plt.savefig(output_file, dpi=120)
plt.close()
def demo():
# 1. 加载数据 (含异常)
raw_df = DataLoader.build_sample_data(seed=7)
print("--- 原始数据 (含计划量为0的返工工单) ---")
print(raw_df[raw_df["order_type"] == "R"].head())
print(f"原始返工工单计划总量: {raw_df[raw_df['order_type']=='R']['planned_qty'].sum():.0f}")
# 2. 计算 (应用修正策略)
calculator = AchievementCalculator(correction_strategy="actual_as_planned")
report = calculator.calculate(raw_df)
# 3. 输出报告
print("\n" + report.summary())
# 4. 可视化
AchievementVisualizer.plot(report, "achievement_report.png")
print("\n📊 达成率报表图已保存:achievement_report.png")
# 5. 工单关系图
G = OrderTypeGraph.build()
OrderTypeGraph.plot(G, "work_order_network.png")
print("📊 工单关系图已保存:work_order_network.png")
# 验证:修正后返工工单计划量应等于实际量
rework_row = report.type_summary[report.type_summary["order_type"] == "R"].iloc[0]
assert abs(rework_row["planned_qty"] - rework_row["actual_qty"]) < 1e-9
print("\n✅ 验证通过:返工工单计划量已修正为实际量。")
return report
if __name__ == "__main__":
demo()
</details>
<details>
<summary></summary>
"""单元测试:工单达成率计算(8/8 通过)。"""
import os
import sys
sys.path.insert(0, os.path.dirname(__file__))
import numpy as np
import pandas as pd
from work_order_achievement import ( # noqa: E402
DataLoader, AchievementCalculator, WorkOrder
)
def _base_df():
return pd.DataFrame({
"order_id": ["N001", "N002"],
"date": ["2026-09-13", "2026-09-13"],
"order_type": ["N", "N"],
"planned_qty": [100, 200],
"actual_qty": [90, 180],
})
def test_normal_achievement_calculation():
df = _base_df()
calc = AchievementCalculator()
report = calc.calculate(df)
# (90+180)/(100+200) = 270/300 = 90%
assert abs(report.total_achievement_rate - 90.0) < 1e-9
print("[PASS] test_normal_achievement_calculation")
def test_rework_zero_planned_correction():
df = _base_df()
# 添加返工工单: 计划0, 实际50
rework = pd.DataFrame([{
"order_id": "R001", "date": "2026-09-13", "order_type": "R",
"planned_qty": 0.0, "actual_qty": 50.0
}])
df = pd.concat([df, rework], ignore_index=True)
calc = AchievementCalculator(correction_strategy="actual_as_planned")
report = calc.calculate(df)
# 修正后: (90+180+50)/(100+200+50) = 320/350 ≈ 91.43%
assert abs(report.total_achievement_rate - (320/350*100)) < 1e-9
# 验证返工工单的计划量已被修正
rework_stat = report.type_summary[report.type_summary["order_type"] == "R"].iloc[0]
assert rework_stat["planned_qty"] == 50.0
print("[PASS] test_rework_zero_planned_correction")
def test_rework_zero_planned_drop_strategy():
df = _base_df()
rework = pd.DataFrame([{
"order_id": "R001", "date": "2026-09-13", "order_type": "R",
"planned_qty": 0.0, "actual_qty": 50.0
}])
df = pd.concat([df, rework], ignore_index=True)
calc = AchievementCalculator(correction_strategy="drop")
report = calc.calculate(df)
# 删除后: (90+180)/(100+200) = 270/300 = 90%
assert abs(report.total_achievement_rate - 90.0) < 1e-9
# 验证返工工单已被删除
assert "R" not in report.type_summary["order_type"].values
print("[PASS] test_rework_zero_planned_drop_strategy")
def test_sample_order_low_achievement():
df = _base_df()
sample = pd.DataFrame([{
"order_id": "S001", "date": "2026-09-13", "order_type": "S",
"planned_qty": 100.0, "actual_qty": 30.0
}])
df = pd.concat([df, sample], ignore_index=True)
calc = AchievementCalculator()
report = calc.calculate(df)
sample_stat = report.type_summary[report.type_summary["order_type"] == "S"].iloc[0]
assert abs(sample_stat["achievement_rate"] - 30.0) < 1e-9
print("[PASS] test_sample_order_low_achievement")
def test_daily_aggregation():
df = pd.DataFrame({
"order_id": ["A", "B", "C"],
"date": ["D1", "D1", "D2"],
"order_type": ["N", "N", "N"],
"planned_qty": [100, 50, 200],
"actual_qty": [80, 40, 180],
})
calc = AchievementCalculator()
report = calc.calculate(df)
assert len(report.daily_summary) == 2
d1 = report.daily_summary[report.daily_summary["date"] == "D1"].iloc[0]
assert d1["planned_qty"] == 150 and d1["actual_qty"] == 120
print("[PASS] test_daily_aggregation")
def test_type_aggregation():
df = pd.DataFrame({
"order_id": ["A", "B", "C"],
"date": ["D1", "D1", "D2"],
"order_type": ["N", "R", "N"],
"planned_qty": [100, 50, 200],
"actual_qty": [80, 40, 180],
})
calc = AchievementCalculator(correction_strategy="actual_as_planned")
report = calc.calculate(df)
assert len(report.type_summary) == 2
n_stat = report.type_summary[report.type_summary["order_type"] == "N"].iloc[0]
assert n_stat["planned_qty"] == 300
print("[PASS] test_type_aggregation")
def test_missing_columns_raises():
df = _base_df().drop(columns=["order_id"])
try:
DataLoader(df=df)
except ValueError as e:
assert "order_id" in str(e)
print("[PASS] test_missing_columns_raises")
return
assert False
def test_empty_after_preprocess():
df = pd.DataFrame({
"order_id": ["X"], "date": ["D1"], "order_type": ["N"],
"planned_qty": 0, "actual_qty": 0
})
calc = AchievementCalculator()
report = calc.calculate(df)
assert report.total_planned == 0
assert report.total_actual == 0
print("[PASS] test_empty_after_preprocess")
if __name__ == "__main__":
for t in [test_normal_achievement_calculation,
test_rework_zero_planned_correction,
test_rework_zero_planned_drop_strategy,
test_sample_order_low_achievement,
test_daily_aggregation,
test_type_aggregation,
test_missing_columns_raises,
test_empty_after_preprocess]:
t()
print("\n全部测试通过 ✅")
</details>
3.3 运行结果(实测 + 手算校验)
--- 原始数据 (含计划量为0的返工工单) ---
order_id date order_type planned_qty actual_qty
3 R001 2026-09-10 R 0.0 38.0
7 R002 2026-09-11 R 0.0 36.0
11 R003 2026-09-12 R 0.0 38.0
15 R004 2026-09-13 R 0.0 34.0
19 R005 2026-09-14 R 0.0 46.0
原始返工工单计划总量: 0
===== 工单达成率报告 =====
总计划产量: 2603
总实际产量: 2504
总达成率: 96.2%
--- 按类型统计 ---
N: 计划=2395, 实际=2328, 达成率=97.2%
R: 计划=192, 实际=192, 达成率=100.0%
S: 计划=16, 实际=15, 达成率=93.8%
📊 达成率报表图已保存:achievement_report.png
📊 工单关系图已保存:work_order_network.png
✅ 验证通过:返工工单计划量已修正为实际量。
手算校验 ✅:
- 返工工单修正:原始
"planned_qty=0",修正后
"planned_qty = actual_qty = 38+36+38+34+46 = 192" ✓
- 总量计算:正常
"2395" + 返工
"192" + 试样
"16" =
"2603" ✓
- 达成率:
"2504 / 2603 * 100 ≈ 96.2%" ✓
- 返工达成率:
"192 / 192 * 100 = 100%"(修正后逻辑自洽)✓
- 单元测试:8/8 通过,PEP8 零告警。
四、README 使用说明
4.1 技术栈(严格限制)
pandas >= 1.5 # 核心:分组聚合 (groupby, agg)
numpy >= 1.24 # 数值计算
matplotlib >= 3.7 # 可视化
networkx >= 3.0 # 工单关系图建模
# 无 sklearn / 无深度学习 / 无数据库依赖
4.2 快速上手
pip install pandas numpy matplotlib networkx
python visualize.py # 生成 achievement_report.png 和 work_order_network.png
python test_work_order_achievement.py # 运行 8 项单元测试
python pack.py # 打包项目
4.3 接入 MES / ERP
from work_order_achievement import DataLoader, AchievementCalculator
# 1. 从 MES 数据库/接口加载数据
# 假设 df 是从 SQL 查询得到的 DataFrame
# df = pd.read_sql("SELECT ... FROM work_orders", connection)
df = DataLoader.build_sample_data() # 演示用
# 2. 初始化计算器 (推荐 'actual_as_planned' 策略)
calculator = AchievementCalculator(correction_strategy="actual_as_planned")
# 3. 计算并获取报告
report = calculator.calculate(df)
# 4. 输出结果 (可写入数据库或发送至 BI 平台)
print(report.summary())
# 将 report.daily_summary 和 report.type_summary 存入数据仓库
# 将 report 对象序列化为 JSON 供前端调用
4.4 关键配置:修正策略
策略 行为 适用场景
"actual_as_planned" 计划=实际(本篇默认) 推荐。保守策略,确保分母不为 0,反映真实消耗。
"drop" 直接删除计划<=0的工单 激进。会丢失返工/试样数据,导致总产量统计偏低。
4.5 扩展方向
方向 说明
加权达成率 根据产品价值/工时,对
"planned_qty" 和
"actual_qty" 进行加权
动态修正 结合历史返工率,预测返工工单的计划量,而非简单设为实际量
实时计算 接入 Kafka/Spark Streaming,实现分钟级达成率监控
根因分析 关联设备停机、质量缺陷数据,分析达成率低的深层原因
五、可视化结果
1. 每日工单达成率报表
- 上图:柱状图对比每日计划产量与实际产量,直观展示产能缺口。
- 下图:折线图展示达成率趋势,100% 基准线辅助判断。
[图片] achievement_report.png
2. 工单类型关系图(NetworkX)
- 展示正常工单(N)可能衍生出返工工单(R)的业务逻辑。
- 解释了为什么返工工单的“计划产量”不能为 0——它是正常生产流程的潜在分支。
[图片] work_order_network.png
六、核心知识点卡片
📌 卡片1:计划产量为 0 是数据模型的“逻辑黑洞”
数学上: 实际产量 / 0 → 无意义 (Infinity)
工程上: 系统强行处理为 1.0 (100%) → 严重误导
课程映射: 智能生产—生产计划与调度
口诀: "无计划,不生产;有计划,才有率"
📌 卡片2:返工工单的“计划”来自哪里?
错误认知: 返工是意外的,所以没计划。
正确认知: 返工是正常质量的"影子计划"。
修正策略: actual_as_planned (计划=实际)
-> 承认返工消耗,修正总达成率。
课程映射: 智能制造技术基础—制造运行管理 (MOM)
口诀: "影子计划不入账,总账永远算不清"
📌 卡片3:pandas 分组聚合是生产统计的“瑞士军刀”
核心代码:
df.groupby(['date', 'order_type']).agg({
'planned_qty': '
利用AI解决实际问题,如果你觉得这个工具好用,欢迎关注长安牧笛!