服饰流行周期计算程序，输入单品上线数据，预判款式衰退清仓最佳时间点。
服饰流行周期计算程序 —— 预判款式衰退与清仓最佳时间点
一、实际应用场景描述
在《时尚产业与品牌创新》课程中，服饰流行周期（Fashion Cycle） 是一个核心理论框架。从 "引入期 → 成长期 → 成熟期 → 衰退期 → 滞销期" 的完整生命周期，决定了品牌每一个关键决策的时间窗口：
阶段 典型表现 品牌动作
引入期 (Introduction) 首批 KOL 穿搭曝光，小红书/抖音种草开始 小批量试水，观察数据
成长期 (Growth) 搜索量攀升，竞品跟进，销量周环比 > 15% 追单补货，加大投放
成熟期 (Maturity) 销量 plateau，周环比 ±5% 波动 维持曝光，准备新品衔接
衰退期 (Decline) 销量周环比持续下滑 > 3 周，退货率上升 启动折扣，减少新流量投入
滞销期 (Obsolescence) 库存周转率 < 0.5，仓储成本 > 边际贡献 清仓退出，回收现金流
核心决策问题：一款衣服上架后，什么时候开始打折？什么时候必须清仓？晚一周可能多赚，但也多压一周库存；早一周可能损失利润。这个"最佳清仓时间点"就是本程序要解决的核心问题。
二、引入痛点
2.1 行业现状的"三拍决策"
痛点 真实表现 后果
拍脑袋定折扣 "卖了一个月了，该打折了吧" 要么打折太早损失毛利，要么太晚变死库存
看总量不看趋势 只看累计销量，不看周环比斜率 成熟期误判为衰退，或衰退期误判为季节性波动
统一折扣节奏 所有款一起 618 打折、双 11 打折 不同款的生命周期阶段完全不同，一刀切效率低
缺乏量化预警 等发现"卖不动了"已经晚了 库存积压 → 折价 70% 才能清掉 → 吞噬前期利润
不懂品类差异 连衣裙和羽绒服用同一套节奏 连衣裙衰退以周计，羽绒服以月计
2.2 一个典型损失场景
某独立品牌，一款 ¥899 的碎花连衣裙：
第 1~3 周：周销量 120 → 150 → 180（成长期，✅ 不打折）
第 4~6 周：周销量 185 → 178 → 170（成熟期，🟡 临界区）
第 7~9 周：周销量 152 → 130 → 105（衰退期，❌ 应该打 8 折）
但实际决策：等到第 10 周才打 7 折
结果：库存 200 件 × 折价损失 ¥180/件 = ¥36,000 利润蒸发
核心矛盾：不是"要不要打折"，而是"在哪一周打折，能使总利润最大化"——这就是流行周期量化的价值。
三、核心逻辑讲解
3.1 流行周期曲线模型
本程序采用 修正的 Bass 扩散模型 + 指数衰减叠加，将单款销量拟合为：
周销量(t) = 基础扩散曲线(t) × 季节衰减因子(t) × 随机噪声(t)
其中：
基础扩散曲线（Bass 模型）：
S(t) = p + (q − p) × F(t) − p × F(t)²
p = 创新系数（KOL/种草驱动，约 0.03~0.08）
q = 模仿系数（大众跟风驱动，约 0.15~0.40）
F(t) = 累积采纳比例
季节衰减因子（品类差异）：
连衣裙：快衰减 λ = 0.12（生命周期 8~12 周）
羽绒服：慢衰减 λ = 0.05（生命周期 16~24 周）
T恤： 中衰减 λ = 0.08（生命周期 12~16 周）
衰减公式：e^(−λ × t)
3.2 衰退判定算法
程序通过三重判定确认衰退起点：
判定条件（需同时满足）：
① 连续 N 周（默认 3 周）周环比 < 0
② 最近一周销量 < 峰值 × 衰退阈值（默认 70%）
③ 斜率（线性回归）显著为负（p < 0.05）
3.3 最佳清仓时间点算法
核心思路：比较"继续持有"vs"立即清仓"的期望利润
持有策略（再等 w 周）：
E[利润(w)] = Σ[P(t) × (1−d(t)) × M] − C_hold × w − C_risk
其中：
P(t) = 第 t 周预测销量
d(t) = 第 t 周自然折扣率（季节性加深）
M = 每件边际贡献（售价 × 毛利率 − 单件可变成本）
C_hold = 周库存持有成本（仓储 + 资金占用 + 贬值）
C_risk = 滞销风险成本（概率 × 最终折价损失）
清仓策略（第 T 周立即打折）：
E[利润(T)] = P(T) × (1−d_clear) × M − C_clear_cost
其中：
d_clear = 清仓折扣率（如 0.30 = 7 折）
C_clear_cost = 清仓执行成本（额外推广/直播坑位费）
最优清仓周 = argmax E[利润(w)] over all w
3.4 折扣策略建议
衰退期折扣阶梯（动态定价）：
衰退早期（销量降至峰值 70%）： 9 折（轻微刺激）
衰退中期（销量降至峰值 50%）： 8 折（主动去库存）
衰退晚期（销量降至峰值 30%）： 6.5~7 折（清仓价）
滞销期 （库存周转率 < 0.3）： 5 折以下（回收现金流）
四、项目结构
fashion_cycle_tracker/
├── config.py # 品类参数配置（衰减率/周期阶段阈值）
├── data_models.py # 数据模型（单品/周销量/生命周期事件）
├── curve_fitter.py # 曲线拟合器（Bass模型 + 季节衰减）
├── decline_detector.py # 衰退检测器（三重判定算法）
├── clearance_optimizer.py # 清仓优化器（利润最大化）
├── discount_strategy.py # 折扣策略生成器
├── report.py # 报告生成（表格 + 可视化）
├── main.py # 主程序入口（含完整示例数据）
├── README.md # 项目说明
└── requirements.txt # 依赖声明
五、代码模块化实现
"requirements.txt"
numpy>=1.24.0
scipy>=1.10.0
matplotlib>=3.7.0
"config.py"
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
config.py
品类参数配置中心：不同服饰品类的生命周期特征参数
"""
from typing import Dict
import numpy as np
# ========== 品类生命周期参数 ==========
# 基于时尚产业研究的典型值
CATEGORY_PARAMS = {
# 连衣裙：春夏核心品类，生命周期短，衰减快
"dress": {
"bass_p": 0.06, # 创新系数（种草敏感度）
"bass_q": 0.35, # 模仿系数（跟风效应强）
"decay_lambda": 0.12, # 衰减速率（快）
"typical_weeks": 10, # 典型总生命周期（周）
"peak_week_range": (3, 6), # 峰值通常出现在第几周
"seasonal_strength": 0.8, # 季节性强度（高）
},
# 羽绒服：秋冬核心品类，生命周期长，衰减慢
"down_jacket": {
"bass_p": 0.04,
"bass_q": 0.25,
"decay_lambda": 0.05,
"typical_weeks": 20,
"peak_week_range": (5, 10),
"seasonal_strength": 0.9,
},
# T恤/基础款：全年可穿，生命周期中等
"tshirt": {
"bass_p": 0.05,
"bass_q": 0.30,
"decay_lambda": 0.08,
"typical_weeks": 14,
"peak_week_range": (4, 8),
"seasonal_strength": 0.4,
},
# 牛仔裤：经典款，生命周期最长
"jeans": {
"bass_p": 0.03,
"bass_q": 0.20,
"decay_lambda": 0.04,
"typical_weeks": 24,
"peak_week_range": (6, 12),
"seasonal_strength": 0.3,
},
# 运动服/瑜伽裤：功能驱动，衰减中等偏快
"activewear": {
"bass_p": 0.07,
"bass_q": 0.38,
"decay_lambda": 0.10,
"typical_weeks": 12,
"peak_week_range": (3, 7),
"seasonal_strength": 0.5,
},
}
# ========== 衰退判定阈值 ==========
DECLINE_THRESHOLDS = {
"min_consecutive_weeks": 3, # 至少连续 N 周下滑
"peak_ratio_threshold": 0.70, # 销量降至峰值的 X% 以下
"slope_p_value": 0.10, # 斜率显著性阈值
"min_weeks_data": 5, # 最少需要多少周数据才判定
}
# ========== 清仓策略参数 ==========
CLEARANCE_PARAMS = {
"holding_cost_per_unit_week": 8.0, # 每件每周持有成本（仓储+资金）
"risk_discount_rate": 0.15, # 滞销风险折价率
"clearance_execution_cost": 500, # 清仓执行成本（直播/推广）
"discount_tiers": { # 折扣阶梯
0.70: "衰退早期（温和刺激）",
0.50: "衰退中期（主动去库存）",
0.30: "衰退晚期（清仓价）",
0.15: "滞销期（回收现金流）",
},
}
# ========== 可视化配色 ==========
COLORS = {
"introduction": "#4CAF50", # 绿 - 引入期
"growth": "#2196F3", # 蓝 - 成长期
"maturity": "#FF9800", # 橙 - 成熟期
"decline": "#F44336", # 红 - 衰退期
"obsolescence": "#9C27B0", # 紫 - 滞销期
"predicted": "#BDBDBD", # 灰 - 预测区间
"clearance": "#FF5722", # 深橙 - 清仓点
"peak": "#E91E63", # 粉 - 峰值
}
"data_models.py"
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
data_models.py
数据模型层：单品定义、周销量数据、生命周期事件记录
"""
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
import numpy as np
from datetime import date, timedelta
@dataclass
class Product:
"""
单品定义
"""
product_id: str # 商品编号，如 "DR-2024-001"
name: str # 商品名称
category: str # 品类（dress/down_jacket/tshirt/jeans/activewear）
price: float # 售价（元）
gross_margin: float # 毛利率（0~1）
launch_date: date # 上架日期
initial_stock: int # 初始库存（件）
variable_cost_per_unit: float = 0.0 # 单件可变成本（包装/物流等）
def __post_init__(self):
if self.category not in CATEGORY_PARAMS:
raise ValueError(f"未注册的品类: {self.category}")
if self.price <= 0:
raise ValueError("售价必须大于零")
if not (0 < self.gross_margin <= 1):
raise ValueError("毛利率必须在 0~1 之间")
def marginal_contribution(self) -> float:
"""每件边际贡献 = 售价 × 毛利率 − 单件可变成本"""
return self.price * self.gross_margin - self.variable_cost_per_unit
@dataclass
class WeeklySales:
"""
周销量数据（支持逐步录入，模拟真实追踪场景）
"""
product_id: str
sales: Dict[int, int] = field(default_factory=dict) # {周序号: 销量}
weeks_recorded: int = 0
def add_week(self, week_num: int, units_sold: int) -> None:
"""录入一周销量"""
if units_sold < 0:
raise ValueError("销量不能为负")
self.sales[week_num] = units_sold
self.weeks_recorded = max(self.weeks_recorded, week_num + 1)
def get_array(self, max_weeks: Optional[int] = None) -> np.ndarray:
"""转为连续数组，缺失周填 0"""
if max_weeks is None:
max_weeks = self.weeks_recorded
arr = np.zeros(max_weeks)
for w, v in self.sales.items():
if w < max_weeks:
arr[w] = v
return arr
def get_peak(self) -> Tuple[int, int]:
"""返回 (峰值周序号, 峰值销量)"""
if not self.sales:
return (-1, 0)
peak_w = max(self.sales, key=self.sales.get)
return (peak_w, self.sales[peak_w])
def week_over_week(self) -> np.ndarray:
"""计算周环比增长率"""
arr = self.get_array()
if len(arr) < 2:
return np.array([])
return (arr[1:] - arr[:-1]) / np.where(arr[:-1] > 0, arr[:-1], 1)
@dataclass
class LifecycleEvent:
"""
生命周期事件记录
"""
event_type: str # introduction/growth/maturity/decline/obsolescence/clearance
week: int # 发生周次
description: str = ""
metadata: Dict = field(default_factory=dict)
"curve_fitter.py"
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
curve_fitter.py
曲线拟合器：用 Bass 扩散模型 + 指数衰减拟合/预测销量曲线
"""
import numpy as np
from scipy.optimize import curve_fit
from typing import Dict, Tuple, Optional
from config import CATEGORY_PARAMS
from data_models import Product, WeeklySales
class BassCurveFitter:
"""
Bass 扩散模型拟合器
核心公式：
S(t) = (p + q × F(t)) × (1 − F(t))
F(t) = 1 − e^(−(p+q)×t) / (1 + (q/p) × e^(−(p+q)×t))
叠加季节衰减：
S_final(t) = S(t) × e^(−λ×t)
"""
def __init__(self, product: Product, weekly_sales: WeeklySales):
self.product = product
self.weekly_sales = weekly_sales
self.category_params = CATEGORY_PARAMS[product.category]
# 拟合结果缓存
self.fitted_params: Optional[Tuple[float, float, float]] = None
self.fitted_curve: Optional[np.ndarray] = None
@staticmethod
def bass_function(t: np.ndarray, p: float, q: float, m: float) -> np.ndarray:
"""
Bass 模型核心函数
t: 时间数组
p: 创新系数
q: 模仿系数
m: 市场总潜力（最大累计采纳数）
"""
# 累计采纳比例 F(t)
if p + q == 0:
f_t = np.zeros_like(t, dtype=float)
else:
exp_term = np.exp(-(p + q) * t)
f_t = (1 - exp_term) / (1 + (q / p) * exp_term) if p > 0 else np.zeros_like(t)
# 当期采纳数（即销量）
s_t = m * (p + q * f_t) * (1 - f_t)
return s_t
def seasonal_decay(self, t: np.ndarray) -> np.ndarray:
"""季节衰减因子"""
lam = self.category_params["decay_lambda"]
strength = self.category_params["seasonal_strength"]
# 基础衰减
decay = np.exp(-lam * t)
# 叠加周期性波动（模拟时尚周期的"长尾"）
seasonal = 1 + 0.15 * strength * np.sin(2 * np.pi * t / 52)
return decay * seasonal
def full_model(self, t: np.ndarray, p: float, q: float, m: float) -> np.ndarray:
"""完整模型 = Bass + 季节衰减"""
bass = self.bass_function(t, p, q, m)
decay = self.seasonal_decay(t)
return bass * decay
def fit(self, max_weeks: int = 30) -> Dict:
"""
用已有数据拟合模型参数
Returns:
拟合结果字典，包含参数、拟合曲线、预测曲线
"""
observed = self.weekly_sales.get_array(max_weeks)
t = np.arange(len(observed))
if len(observed) < 4:
raise ValueError("数据不足 4 周，无法拟合")
# 初始参数猜测
p0 = self.category_params["bass_p"]
q0 = self.category_params["bass_q"]
m0 = observed.max() * 3 # 市场潜力约为峰值的 3 倍
try:
popt, pcov = curve_fit(
self.full_model,
t,
observed,
p0=[p0, q0, m0],
bounds=([0, 0, 0], [1, 1, 1e6]),
maxfev=5000,
)
self.fitted_params = tuple(popt)
# 生成拟合曲线（已有数据部分）
self.fitted_curve = self.full_model(t, *popt)
# 预测未来（再预测 max_weeks 周）
t_future = np.arange(max_weeks)
future_curve = self.full_model(t_future, *popt)
return {
"status": "success",
"params": {
"p (创新系数)": round(popt[0], 4),
"q (模仿系数)": round(popt[1], 4),
"m (市场潜力)": round(popt[2], 1),
},
"fitted_weeks": len(observed),
"fitted_curve": self.fitted_curve,
"predicted_curve": future_curve,
"predicted_weeks": max_weeks,
"rss": float(np.sum((observed - self.fitted_curve) ** 2)),
}
except Exception as e:
return {
"status": "failed",
"error": str(e),
"params": {"p": p0, "q": q0, "m": m0},
}
def predict_future(self, total_weeks: int = 20) -> np.ndarray:
"""预测未来销量"""
if self.fitted_params is None:
raise ValueError("请先调用 fit() 拟合模型")
t = np.arange(total_weeks)
return self.full_model(t, *self.fitted_params)
def identify_stage(self, week: int) -> str:
"""
根据拟合曲线判断某周所处的生命周期阶段
"""
if self.fitted_params is None:
return "unknown"
t = np.arange(max(week + 1, 10))
curve = self.full_model(t, *self.fitted_params)
if week >= len(curve):
return "obsolescence"
# 找到峰值位置
peak_idx = int(np.argmax(curve))
peak_val = curve[peak_idx]
current_val = curve[week]
peak_ratio = current_val / peak_val if peak_val > 0 else 0
if week < peak_idx * 0.5:
return "introduction"
elif week < peak_idx:
return "growth"
elif peak_ratio > 0.70:
return "maturity"
elif peak_ratio > 0.35:
return "decline"
else:
return "obsolescence"
"decline_detector.py"
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
decline_detector.py
衰退检测器：三重判定算法识别款式进入衰退期的精确时间点
"""
import numpy as np
from typing import Dict, List, Tuple
from scipy import stats
from config import DECLINE_THRESHOLDS
from data_models import WeeklySales
class DeclineDetector:
"""
衰退检测算法
三重判定：
① 连续 N 周销量环比下滑
② 销量降至峰值 X% 以下
③ 线性回归斜率显著为负
"""
def __init__(self, weekly_sales: WeeklySales):
self.sales = weekly_sales
self.thresholds = DECLINE_THRESHOLDS
def check_consecutive_decline(self, window: int = 3) -> Dict:
"""
判定①：检查是否存在连续 N 周下滑
Returns:
{detected: bool, start_week: int, decline_weeks: int}
"""
arr = self.sales.get_array()
if len(arr) < window:
return {"detected": False, "reason": "数据不足"}
# 滑动窗口检测
max_consecutive = 0
current_consecutive = 0
decline_start = -1
for i in range(1, len(arr)):
if arr[i] < arr[i - 1]:
current_consecutive += 1
if current_consecutive == 1:
decline_start = i - 1
max_consecutive = max(max_consecutive, current_consecutive)
else:
current_consecutive = 0
min_required = self.thresholds["min_consecutive_weeks"]
detected = max_consecutive >= min_required
return {
"detected": detected,
"max_consecutive_decline": max_consecutive,
"decline_start_week": decline_start if detected else -1,
"reason": f"最长连续下滑 {max_consecutive} 周" if detected else "未检测到连续下滑",
}
def check_peak_ratio(self) -> Dict:
"""
判定②：当前销量是否降至峰值的 X% 以下
Returns:
{detected: bool, current_ratio: float, peak_week: int, peak_value: int}
"""
peak_week, peak_val = self.sales.get_peak()
if peak_week < 0:
return {"detected": False, "reason": "无销量数据"}
arr = self.sales.get_array()
threshold = self.thresholds["peak_ratio_threshold"]
# 检查最近几周是否低于阈值
recent_weeks = 3
detected = False
for i in range(max(0, len(arr) - recent_weeks), len(arr)):
ratio = arr[i] / peak_val if peak_val > 0 else 0
if ratio < threshold:
detected = True
break
current_ratio = arr[-1] / peak_val if peak_val > 0 and len(arr) > 0 else 0
return {
"detected": detected,
"current_ratio": round(current_ratio, 3),
"peak_week": peak_week,
"peak_value": peak_val,
"threshold": threshold,
"reason": f"当前销量是峰值的 {current_ratio*100:.1f}%" if detected else "仍在峰值以上",
}
def check_slope_significance(self, window: int = 5) -> Dict:
"""
判定③：线性回归斜率是否显著为负
Returns:
{detected: bool, slope: float, p_value: float}
"""
arr = self.sales.get_array()
if len(arr) < window:
return {"detected": False, "reason": "数据不足"}
# 取最近 window 周做线性回归
recent = arr[-window:]
t = np.arange(window)
slope, intercept, r_value, p_value, std_err = stats.linregress(t, recent)
alpha = self.thresholds["slope_p_value"]
detected = slope < 0 and p_value < alpha
return {
"detected": detected,
"slope": round(slope, 2),
"r_squared": round(r_value ** 2, 3),
"p_value": round(p_value, 4),
"reason": f"斜率={slope:.1f}, p={p_value:.3f}" if detected else "斜率不显著",
}
def detect_decline(self) -> Dict:
"""
三重判定汇总
Returns:
综合判定结果，包含各子判定详情和最终结论
"""
if self.sales.weeks_recorded < self.thresholds["min_weeks_data"]:
return {
"status": "insufficient_data",
"weeks_recorded": self.sales.weeks_recorded,
"min_required": self.thresholds["min_weeks_data"],
"decline_detected": False,
}
result_1 = self.check_consecutive_decline()
result_2 = self.check_peak_ratio()
result_3 = self.check_slope_significance()
# 三重判定汇总
count = sum([
result_1["detected"],
result_2["detected"],
result_3["detected"],
])
# 至少满足 2/3 才判定为衰退
decline_detected = count >= 2
# 衰退起点 = 最严格的判定（连续下滑起点）
decline_week = -1
if decline_detected:
candidates = []
if result_1["detected"]:
candidates.append(result_1["decline_start_week"])
if result_2["detected"]:
candidates.append(result_2.get("peak_week", -1))
if result_3["detected"]:
candidates.append(self.sales.weeks_recorded - 1)
decline_week = min(c for c in candidates if c >= 0)
return {
"status": "analyzed",
"decline_detected": decline_detected,
"decline_week": decline_week,
"confidence": f"{count}/3 条件满足",
"checks": {
"连续下滑检测": result_1,
"峰值比例检测": result_2,
"斜率显著性检测": result_3,
},
"recommendation": self._get_recommendation(
decline_detected, count, decline_week
),
}
def _get_recommendation(
self, detected: bool, count: int, week: int
) -> str:
"""根据判定结果给出建议"""
if not detected:
if count == 0:
return "款式仍处于健康销售期，维持正常运营"
elif count == 1:
return "出现早期衰退信号，建议密切关注后续 1~2 周数据"
else:
return "数据矛盾，建议人工复核"
else:
return f"⚠️ 衰退已确认（第 {week} 周起），建议启动折扣/清仓评估"
class LifecycleTracker:
"""
生命周期阶段追踪器
基于销量曲线的形态特征，自动标注每个阶段
"""
@staticmethod
def identify_stages(sales_array: np.ndarray) -> List[Tuple[int, str]]:
"""
基于销量曲线的二阶导数识别阶段转换点
Returns:
[(week, stage_name), ...] 按时间排序的阶段列表
"""
if len(sales_array) < 4:
return [(0, "introduction")]
stages = []
arr = sales_array.copy()
# 一阶导数（变化率）
diff1 = np.diff(arr)
# 二阶导数（加速度）
diff2 = np.diff(diff1)
# 找峰值
peak_idx = int(np.argmax(arr))
# 找拐点（二阶导数过零）
inflection_points = []
for i in range(1, len(diff2)):
if diff2[i] * diff2[i - 1] < 0:
inflection_points.append(i + 1)
# 阶段划分逻辑
# 引入期：0 ~ 第一个拐点（或峰值的 1/3）
intro_end = inflection_points[0] if inflection_points else peak_idx // 3
stages.append((0, "introduction"))
# 成长期：拐点 ~ 峰值
if intro_end < peak_idx:
stages.append((intro_end, "growth"))
# 成熟期：峰值 ~ 峰值后第一个拐点
decline_start = peak_idx + 1
for ip in inflection_points:
if ip > peak_idx:
decline_start = ip
break
stages.append((peak_idx, "maturity"))
# 衰退期：拐点 ~ 低谷
if decline_start < len(arr) - 1:
stages.append((decline_start, "decline"))
利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！