国货回潮消费趋势预测与国风爆款品类预判系统定位文化驱动型时尚消费趋势预测教学工具适用《时尚产业与品牌创新》课程实验、趋势企划、商品企划语言Python 3.8说明采用时间序列 热度传播模型模拟文化热点扩散过程全流程可复现、中立化、无商业引流一、实际应用场景描述在《时尚产业与品牌创新》课程中一个前沿课题是国货回潮不是匀速的——它受文化热点驱动有明显的波段特征。如何用数据预判下一波国风爆款品类近年来典型文化热点与爆款联动案例文化热点 时间节点 爆款品类 代表品牌河南卫视《唐宫夜宴》 2021 Q1 唐制汉服、襦裙 十三余、兰若庭北京冬奥会中国红 2022 Q1 红色系运动服、国潮卫衣 安踏、李宁马面裙街拍出圈 2023 Q2-Q3 马面裙改良、百迭裙 各类新中式品牌电视剧《长相思》热播 2023 Q3 战国风配饰、玉佩元素 饰品品牌甲辰龙年生肖 2024 Q1 龙纹刺绣、龙年联名 周大福、茶颜悦色品牌面临的决策痛点- 文化热点来了但反应慢半拍→ 爆款没赶上- 押注了某个文化主题但热度没起来→ 库存积压- 不知道下一个爆款品类是什么→ 企划靠猜二、引入痛点 痛点 1文化热点 ≠ 消费转化现象 问题故宫文创火了我们也做 没算热度衰减曲线马面裙突然爆了 事后归因事前无预警龙年做龙纹 同质化严重无差异化 文化热度到消费转化有时间滞后 衰减效应 痛点 2缺乏量化预判工具- 趋势报告滞后 1-2 个季度- 社交媒体监听工具贵且不准- 课程作业缺乏可操作的时间序列预测模型 痛点 3品类优先级模糊问题 后果汉服、马面裙、新中式、国潮运动服……都做 资源分散哪个品类先爆 无法排序热点持续多久 备货量拍脑袋三、核心逻辑讲解✅ 文化热点传播模型SIR 简化版文化热点在人群中的传播类似传染病模型S (Susceptible) → I (Infected/传播者) → R (Recovered/遗忘)热度峰值 ≈ 传播率 β / 恢复率 γ 的比值转化为消费热度品类热度(t) 基础热度 文化驱动力(t) × 品类关联度 × 衰减因子✅ 品类爆款预判公式爆款得分 w₁ × 文化关联度 w₂ × 搜索趋势斜率一阶导数 w₃ × 社交声量归一化 w₄ × 品类成熟度越高越稳- w₅ × 竞争饱和度✅ 季度预判流程输入当前季度文化热点列表│├─ Step 1: 计算每个热点的传播曲线SIR 模型├─ Step 2: 计算各品类与热点的关联度矩阵├─ Step 3: 叠加多热点影响 → 品类热度预测├─ Step 4: 排序 → 爆款品类 TOP N└─ Step 5: 输出品类预警何时入场 / 何时退场四、代码模块化实现 项目结构guohui_trend_forecaster/│├── config.py # 文化热点、品类、权重参数├── sir_model.py # SIR 文化传播模型├── trend_analyzer.py # 热度趋势分析斜率、加速度├── category_scorer.py # 品类爆款得分计算├── forecaster.py # 季度预判主引擎├── visualizer.py # 传播曲线、热度堆叠图├── reporter.py # 品类预警报告├── main.py # 入口└── README.mdconfig.py# config.pyfrom dataclasses import dataclass, fieldfrom typing import Dict, List# 文化热点定义 dataclassclass CulturalEvent:文化热点事件name: strquarter: str # 发生季度如 2025-Q2peak_week: int # 爆发后第几周达峰值1-52base_intensity: float # 基础传播强度 (0-1)cultural_theme: str # 文化主题标签related_keywords: List[str] field(default_factorylist)# 2025 年预设文化热点 CULTURAL_EVENTS_2025 [CulturalEvent(name春节非遗民俗季,quarter2025-Q1,peak_week4,base_intensity0.85,cultural_theme非遗/民俗,related_keywords[剪纸, 皮影, 泥塑, 年画]),CulturalEvent(name清明节·国风踏青,quarter2025-Q2,peak_week14,base_intensity0.45,cultural_theme节气/自然,related_keywords[踏青, 汉服出游, 清明]),CulturalEvent(name端午节·龙舟文化,quarter2025-Q2,peak_week22,base_intensity0.60,cultural_theme传统节日,related_keywords[龙舟, 粽子, 艾草, 五彩绳]),CulturalEvent(name暑期古装剧热播季,quarter2025-Q3,peak_week30,base_intensity0.75,cultural_theme影视IP,related_keywords[古装剧, 国风配饰, 发簪, 团扇]),CulturalEvent(name中秋节·国风浪漫,quarter2025-Q3,peak_week38,base_intensity0.55,cultural_theme传统节日,related_keywords[嫦娥, 玉兔, 桂花, 圆月]),CulturalEvent(name国庆·文化自信周,quarter2025-Q4,peak_week42,base_intensity0.80,cultural_theme国家叙事,related_keywords[中国红, 国徽元素, 华表, 长城]),]# 国风品类定义 dataclassclass Category:name: strbase_popularity: float # 基础热度 (0-1)maturity: float # 品类成熟度 (0-1越高越稳定)competition: float # 竞争饱和度 (0-1)avg_price_point: int # 核心价格带元# 与各文化主题的关联度 (0-1)theme_relevance: Dict[str, float] field(default_factorydict)CATEGORIES [Category(马面裙改良, 0.75, 0.70, 0.65, 399, {非遗/民俗: 0.85, 传统节日: 0.60, 影视IP: 0.55,节气/自然: 0.50, 国家叙事: 0.40,}),Category(新中式上衣盘扣/立领, 0.70, 0.65, 0.55, 299, {非遗/民俗: 0.80, 传统节日: 0.70, 影视IP: 0.65,节气/自然: 0.45, 国家叙事: 0.50,}),Category(国潮运动服, 0.65, 0.80, 0.75, 499, {国家叙事: 0.90, 传统节日: 0.50, 非遗/民俗: 0.40,影视IP: 0.35, 节气/自然: 0.30,}),Category(汉服日常化, 0.60, 0.55, 0.45, 599, {非遗/民俗: 0.90, 传统节日: 0.85, 影视IP: 0.80,节气/自然: 0.70, 国家叙事: 0.35,}),Category(国风配饰发簪/团扇/玉佩, 0.55, 0.50, 0.40, 159, {影视IP: 0.85, 传统节日: 0.75, 非遗/民俗: 0.70,节气/自然: 0.60, 国家叙事: 0.30,}),Category(刺绣单品苏绣/粤绣元素, 0.50, 0.45, 0.35, 799, {非遗/民俗: 0.95, 影视IP: 0.60, 传统节日: 0.55,节气/自然: 0.40, 国家叙事: 0.35,}),Category(扎染/蜡染服饰, 0.40, 0.35, 0.25, 459, {非遗/民俗: 0.90, 节气/自然: 0.75, 传统节日: 0.50,影视IP: 0.40, 国家叙事: 0.20,}),Category(唐制/明制汉服, 0.45, 0.40, 0.50, 1299, {非遗/民俗: 0.85, 传统节日: 0.80, 影视IP: 0.75,节气/自然: 0.55, 国家叙事: 0.30,}),]# 评分权重 dataclassclass ScoringWeights:theme_relevance: float 0.35 # 文化关联度trend_slope: float 0.25 # 搜索趋势斜率social_volume: float 0.20 # 社交声量maturity: float 0.10 # 品类成熟度稳定性competition_penalty: float 0.10 # 竞争惩罚# SIR 模型参数 dataclassclass SIRParams:beta: float 0.35 # 传播率gamma: float 0.08 # 恢复率遗忘率noise_level: float 0.05total_weeks: int 52# 输出文件 OUTPUT_DIR trend_forecastsir_model.py# sir_model.pySIR 文化热点传播模型模拟文化热点从爆发 → 扩散 → 衰减的全过程import numpy as npimport pandas as pdfrom config import SIRParams, CulturalEventdef sir_simulation(event: CulturalEvent, params: SIRParams None) - pd.DataFrame:对单个文化热点运行 SIR 模型返回每周热度 DataFrameweek, S, I, R, heat_indexif params is None:params SIRParams()N 1.0 # 总人群归一化I0 event.base_intensity * 0.15 # 初始感染者种子用户比例S0 1.0 - I0R0 0.0np.random.seed(42)records []S, I, R S0, I0, R0for t in range(params.total_weeks):# SIR 微分方程欧拉法离散化dS -params.beta * S * IdI params.beta * S * I - params.gamma * IdR params.gamma * IS max(0, S dS)I max(0, I dI)R min(1, R dR)# 热度指数 感染者比例 恢复者比例 × 衰减heat I R * np.exp(-0.05 * t)# 叠加随机噪声noise np.random.normal(0, params.noise_level)heat max(0, min(1, heat noise))records.append({week: t 1,S: round(S, 4),I: round(I, 4),R: round(R, 4),heat_index: round(heat, 4),event: event.name,})df pd.DataFrame(records)# 标记峰值周peak_w df[heat_index].idxmax() 1df[is_peak] df.index (peak_w - 1)return dfdef batch_simulate(events: list[CulturalEvent]) - dict[str, pd.DataFrame]:批量模拟多个文化热点results {}for evt in events:df sir_simulation(evt)results[evt.name] dfprint(f ✓ 模拟完成: {evt.name}峰值周: {df[heat_index].idxmax() 1})return resultstrend_analyzer.py# trend_analyzer.py热度趋势分析计算斜率一阶导数、加速度二阶导数import numpy as npimport pandas as pdfrom config import CulturalEventdef compute_slope(heat_series: pd.Series, window: int 3) - pd.Series:计算热度变化斜率一阶差分平滑return heat_series.rolling(windowwindow, centerTrue, min_periods1).apply(lambda x: np.gradient(x) if len(x) 1 else 0)def compute_acceleration(slope_series: pd.Series, window: int 3) - pd.Series:计算加速度二阶导数return slope_series.rolling(windowwindow, centerTrue, min_periods1).apply(lambda x: np.gradient(x) if len(x) 1 else 0)def analyze_event_trend(event: CulturalEvent, heat_df: pd.DataFrame) - dict:分析单个热点的趋势特征heat heat_df[heat_index]slope compute_slope(heat)accel compute_acceleration(slope)# 关键指标peak_week heat.idxmax() 1peak_value heat.max()# 爆发力前 4 周的平均斜率early_slope slope[:4].mean()# 持久力峰值后 4 周的热度衰减速度if peak_week len(heat) - 4:decay_slope slope[peak_week:peak_week 4].mean()else:decay_slope 0# 总热度面积积分近似total_heat_area heat.sum()return {event: event.name,peak_week: int(peak_week),peak_value: round(peak_value, 4),early_slope: round(early_slope, 4),decay_slope: round(decay_slope, 4),total_heat_area: round(total_heat_area, 4),heat_slope: slope,heat_acceleration: accel,}def batch_analyze(events: list[CulturalEvent], sim_results: dict) - dict[str, dict]:批量分析results {}for evt in events:df sim_results[evt.name]results[evt.name] analyze_event_trend(evt, df)return resultscategory_scorer.py# category_scorer.py品类爆款得分计算综合文化关联度、趋势斜率、社交声量、成熟度、竞争度import numpy as npimport pandas as pdfrom config import Category, ScoringWeights, CulturalEventdef compute_theme_relevance(cat: Category, event: CulturalEvent) - float:品类与当前热点的文化关联度return cat.theme_relevance.get(event.cultural_theme, 0.1)def compute_social_volume(cat: Category, event: CulturalEvent) - float:模拟社交声量关联度越高 基础热度越高 → 声量越大relevance compute_theme_relevance(cat, event)base cat.base_popularityvolume relevance * 0.6 base * 0.4# 加一点随机性noise np.random.normal(0, 0.05)return max(0, min(1, volume noise))def score_category(cat: Category,event: CulturalEvent,trend_metrics: dict,weights: ScoringWeights None) - dict:计算单个品类在给定热点下的爆款得分if weights is None:weights ScoringWeights()relevance compute_theme_relevance(cat, event)social compute_social_volume(cat, event)# 趋势斜率归一化到 0-1slope_norm max(0, min(1, (trend_metrics[early_slope] 0.5) / 1.0))score (weights.theme_relevance * relevance weights.trend_slope * slope_norm weights.social_volume * social weights.maturity * cat.maturity- weights.competition_penalty * cat.competition)return {category: cat.name,theme_relevance: round(relevance, 3),trend_slope: round(slope_norm, 3),social_volume: round(social, 3),maturity: cat.maturity,competition: cat.competition,raw_score: round(score, 4),avg_price: cat.avg_price_point,}def rank_categories_for_event(categories: list[Category],event: CulturalEvent,trend_metrics: dict,weights: ScoringWeights None) - pd.DataFrame:对单个热点给所有品类打分并排名scores []for cat in categories:s score_category(cat, event, trend_metrics, weights)scores.append(s)df pd.DataFrame(scores)df[rank] df[raw_score].rank(ascendingFalse, methodmin).astype(int)# 归一化得分0-100mn, mx df[raw_score].min(), df[raw_score].max()if mx mn:df[normalized_score] ((df[raw_score] - mn) / (mx - mn) * 100).round(1)else:df[normalized_score] 50.0return df.sort_values(rank).reset_index(dropTrue)forecaster.py# forecaster.py季度预判主引擎输入季度文化热点 → 输出品类爆款预判排名import numpy as npimport pandas as pdfrom config import CULTURAL_EVENTS_2025, CATEGORIES, ScoringWeightsfrom sir_model import batch_simulatefrom trend_analyzer import batch_analyzefrom category_scorer import rank_categories_for_eventdef forecast_quarter(quarter: str,events: list None,categories: list None,weights: ScoringWeights None) - dict:对指定季度进行品类爆款预判返回{quarter: 季度名,events: 该季度热点列表,category_rankings: {event_name: 品类排名 DataFrame},consolidated_ranking: 综合排名 DataFrame,weekly_heat: {event_name: 每周热度 DataFrame},}if events is None:events CULTURAL_EVENTS_2025if categories is None:categories CATEGORIESif weights is None:weights ScoringWeights()# 筛选该季度热点q_events [e for e in events if e.quarter quarter]if not q_events:return {quarter: quarter, error: 该季度无预设文化热点}print(f\n{*60})print(f {quarter} — 文化热点列表:)for e in q_events:print(f • {e.name}主题: {e.cultural_theme})print(f{*60})# Step 1: SIR 模拟print(f\n[Step 1] SIR 文化传播模拟...)sim_results batch_simulate(q_events)# Step 2: 趋势分析print(f\n[Step 2] 热度趋势分析...)trend_results batch_analyze(q_events, sim_results)# Step 3: 品类打分print(f\n[Step 3] 品类爆款得分计算...)all_rankings {}for evt in q_events:df rank_categories_for_event(categories, evt, trend_results[evt.name], weights)all_rankings[evt.name] df# Step 4: 综合排名加权汇总多热点print(f\n[Step 4] 综合品类排名...)consolidated consolidate_rankings(all_rankings, q_events)return {quarter: quarter,events: q_events,category_rankings: all_rankings,consolidated_ranking: consolidated,weekly_heat: sim_results,trend_metrics: trend_results,}def consolidate_rankings(rankings: dict[str, pd.DataFrame],events: list) - pd.DataFrame:多热点下的综合品类排名使用 Borda Count 投票法每个热点下的排名转化为分数all_cats set()for df in rankings.values():all_cats.update(df[category])# 每个热点给品类打分排名越靠前分数越高scores {cat: 0.0 for cat in all_cats}weights {evt.name: evt.base_intensity for evt in events}for evt_name, df in rankings.items():n len(df)for _, row in df.iterrows():# Borda 分数 (n - rank 1) × 热点权重borda (n - row[rank] 1) * weights[evt_name]scores[row[category]] borda# 转 DataFrameresult pd.DataFrame([{category: cat, borda_score: round(s, 2)}for cat, s in scores.items()]).sort_values(borda_score, ascendingFalse).reset_index(dropTrue)result.index result.index 1result.index.name 综合排名return resultvisualizer.py# visualizer.py可视化模块传播曲线、热度堆叠图、品类排名条形图import matplotlib.pyplot as pltimport numpy as npimport pandas as pdplt.rcParams[font.sans-serif] [Noto Sans CJK SC, SimHei, Microsoft YaHei]plt.rcParams[axes.unicode_minus] Falsedef plot_sir_curves(sim_results: dict, save_path: str sir_heat_curves.png):绘制所有热点的 SIR 传播曲线n len(sim_results)fig, axes plt.subplots(2, 3, figsize(18, 10))axes axes.flatten()colors plt.cm.Set1(np.linspace(0, 1, n))for i, (name, df) in enumerate(sim_results.items()):ax axes[i]ax.plot(df[week], df[I], colorcolors[i], linewidth2.5, label传播中 (I))ax.fill_between(df[week], df[I], alpha0.15, colorcolors[i])# 标注峰值peak_idx df[heat_index].idxmax()ax.annotate(f峰值 W{peak_idx1}\n{df[heat_index].max():.2f},xy(peak_idx1, df[heat_index].max()),xytext(peak_idx-5, df[heat_index].max()0.1),fontsize9, hacenter,arrowpropsdict(arrowstyle-, colorgray))ax.set_title(name, fontsize12, fontweightbold)ax.set_xlabel(周次, fontsize10)ax.set_ylabel(热度指数, fontsize10)ax.set_ylim(0, 1.1)ax.grid(True, alpha0.3)# 隐藏多余的子图for j in range(n, len(axes)):axes[j].axis(off)fig.suptitle(文化热点传播曲线SIR 模型, fontsize16, fontweightbold, y1.01)plt.tight_layout()plt.savefig(save_path, dpi150, bbox_inchestight)plt.close()print(f✅ SIR 传播曲线已保存: {save_path})def plot_category_ranking(ranking_df: pd.DataFrame, event_name: str,save_path: str category_ranking.png):品类爆款排名条形图df ranking_df.head(8).copy()df df.iloc[::-1] # 反转使排名 1 在最上面fig, ax plt.subplots(figsize(10, 6))colors plt.cm.YlOrRd(np.linspace(0.3, 0.9, len(df)))bars ax.barh(range(len(df)), df[normalized_score], colorcolors, edgecolorwhite)ax.set_yticks(range(len(df)))ax.set_yticklabels(df[category], fontsize11)ax.set_xlabel(爆款得分归一化 0-100, fontsize12)ax.set_title(f「{event_name}」驱动 → 品类爆款预判排名, fontsize14, fontweightbold)# 标注排名for i, (_, row) in enumerate(df.iterrows()):ax.text(row[normalized_score] 1, i, f#{row[rank]} ({row[normalized_score]:.0f}分),vacenter, fontsize9, colordimgray)plt.tight_layout()plt.savefig(save_path, dpi150, bbox_inchestight)plt.close()print(f✅ 品类排名图已保存: {save_path})def plot_consolidated_ranking(consolidated_df: pd.DataFrame,save_path: str consolidated_ranking.png):综合排名堆叠条形图df consolidated_df.head(8).copy()df df.iloc[::-1]fig, ax plt.subplots(figsize(10, 6))bars ax.barh(range(len(df)), df[borda_score], color#C41E3A, alpha0.85, edgecolorwhite)ax.set_yticks(range(len(df)))ax.set_yticklabels(df[category], fontsize11)ax.set_xlabel(Borda 综合得分, fontsize12)ax.set_title( 季度品类爆款综合预判排名, fontsize15, fontweightbold)for i, (_, row) in enumerate(df.iterrows()):ax.text(row[borda_score] 0.5, i, f#{row.name} ({row[borda_score]:.1f}),vacenter, fontsize10, fontweightbold)plt.tight_layout()plt.savefig(save_path, dpi150, bbox_inchestight)plt.close()print(f✅ 综合排名图已保存: {save_path})reporter.py# reporter.py品类预警报告生成给出每个品类的入场时机和退场预警import pandas as pddef generate_timing_alert(event: object,heat_df: pd.DataFrame,category_rank: pd.DataFrame) - dict:生成品类时间预警peak_week heat_df[heat_index].idxmax() 1# 热度上升期 入场窗口峰值前 2-4 周enter_start max(1, peak_week - 4)enter_end max(1, peak_week - 2)# 热度衰退期 退场预警峰值后 3-6 周exit_start min(len(heat_df), peak_week 3)exit_end min(len(heat_df), peak_week 6)# 各品类预警alerts []for _, row in category_rank.iterrows():rank row[rank]if rank 3:urgency 高优先级action f立即备货W{enter_start}-W{enter_end} 上架elif rank 5:urgency 中优先级action fW{enter_start} 启动企划W{enter_end} 前上架else:urgency ⚪ 低优先级action f观望W{enter_start} 评估是否跟进alerts.append({rank: rank,category: row[category],urgency: urgency,action: action,enter_window: fW{enter_start} ~ W{enter_end},exit_alert: fW{exit_start} ~ W{exit_end},score: row[normalized_score],})return {event: event.name,peak_week: peak_week,peak_value: round(heat_df[heat_index].max(), 3),alerts: alerts,}def print_timing_report(alerts: list[dict]):打印时间预警报告for a in alerts:print(f\n{*65})print(f {a[event]})print(f 热度峰值: W{a[peak_week]}指数 {a[peak_value]})print(f{*65})for item in a[alerts]:print(f\n {item[urgency]} #{item[rank]} {item[category]})print(f 爆款得分: {item[score]})print(f 建议入场: {item[enter_window]})print(f 退场预警: {item[exit_alert]})print(f 行动: {item[ac利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛