媒体种草投放ROI计算器

一、实际应用场景描述

在《时尚产业与品牌创新》课程中，媒体投放ROI（Return on Investment）是衡量营销效率的核心指标。一个典型的品牌营销决策场景： “这季新品，短视频投50万还是杂志投30万？哪个带回更多销量？哪个让单品更赚钱？” 这个问题远比“哪个获赞多”复杂。需要串联的完整链路是： 投放预算 → 内容曝光 → 互动转化 → 电商引流 → 实际成交 → 单品收益核算 ↓ ↓ ↓ ↓ ↓ ↓ 成本中心 流量漏斗 转化漏斗 销售漏斗 收入中心 ROI输出 传统做法的局限：品牌通常用“曝光量”或“互动量”衡量投放效果，但这只是中间指标，不是商业结果。曝光1000万≠卖出1000件。真正该算的是： 每1元投放，带回多少净利润？哪个渠道的净利润最高？哪个单品的投放回报最好？ 本程序的目标：输入短视频和杂志广告的投放预算，结合转化率、客单价、退货率等参数，自动核算每个渠道、每个单品的ROI，输出决策建议。

二、引入痛点

2.1 行业现状问题

痛点具体表现后果：

• 只看表面指标：“这条视频500万播放！”播放量≠销量，虚假繁荣

• 渠道割裂核算：短视频算短视频的，杂志算杂志的，无法横向对比“该投哪个”

• 忽略真实转化：没扣除退货/取消订单，ROI虚高20%~40%

• 单品维度缺失：只算“品牌整体ROI”，不知道“哪件单品最值得投”

• 归因模糊：不知道成交来自哪个渠道，预算分配靠“感觉”

• 非线性成本：忽略“投放越多、边际效用递减”，过度投放，后期效率极低

  2.2 一个典型决策失误场景

  某女装品牌2024春季投放： 决策：短视频投¥80万，时尚杂志投¥50万 结果（只看表面）： 短视频：播放量3000万，点赞60万 → “效果很好！” 杂志：阅读量80万 → “不如短视频” 实际核算（本程序跑出来的）： 短视频：投放¥80万 → 引流2.4万 → 成交4800单 → 退货1440单 实际有效成交 3360 单 × ¥399 = ¥134 万 净收入 - 产品成本 - 投放成本 = 实际 ROI 0.45 杂志：投放 ¥50 万 → 引流 8000 → 成交 2400 单 → 退货 240 单 实际有效成交 2160 单 × ¥799 = ¥173 万 净收入 - 产品成本 - 投放成本 = 实际 ROI 1.8 真相：杂志 ROI 是短视频的 4 倍！ 原因：杂志读者购买意向更强（转化率高 3 倍），退货率低 80% 但品牌因为“短视频播放量好看”，下季又追加了短视频预算…… 核心矛盾：不是“有没有数据”的问题，而是核算维度错了——用“曝光量”做决策，而非“净利润”。 三、核心逻辑讲解 3.1 整体架构 输入层 计算层 输出层 ┌────────────────┐ ┌──────────────────┐ ┌────────────────┐ │ 短视频投放预算 │ │ 流量漏斗模型 │ │ 渠道 ROI 对比 │ │ 杂志广告预算 │ → │ 转化漏斗模型 │ → │ 单品 ROI 排名 │ │ 单品售价/成本 │ │ 归因分配模型 │ │ 边际效用曲线 │ │ 转化率/退货率 │ │ 净利润核算 │ │ 预算优化建议 │ └────────────────┘ └──────────────────┘ └────────────────┘ 3.2 流量漏斗模型 短视频渠道： 投放预算 → 千次曝光成本(CPM) → 总曝光量 → 点击率(CTR) → 点击量（引流到电商） → 转化率(CVR) → 下单量 → 退货率 → 有效成交 杂志渠道： 投放预算 → 千次阅读成本(CPM_mag) → 总阅读量 → 扫码/搜索转化率 → 点击量 → 转化率(CVR_mag) → 下单量 → 退货率 → 有效成交 关键差异参数： 短视频 CTR: 1.5%~5% 杂志 CTR: 0.3%~1.2% 短视频 CVR: 0.5%~3% 杂志 CVR: 3%~8% 短视频退货率: 25%~40% 杂志退货率: 5%~15% 3.3 归因分配模型 当同一个用户既看了短视频又看了杂志广告，成交该算谁的？ 采用时间衰减归因 + 渠道权重： 重构内容： 单次成交的渠道贡献 = Σ(渠道权重_i × 时间衰减因子_i) 渠道权重（默认）：

• 短视频: 0.6（即时性强，冲动消费）

• 杂志: 0.3（信任背书，决策辅助）

• 自然搜索: 0.1（品牌词搜索）

• 时间衰减：接触后第 N 天成交，权重 × 0.8^N 单品收益核算 单品净收益 = 有效成交数 × (客单价 − 产品成本 − 单件物流成本) − 投放成本分摊 投放成本分摊（按引流贡献比例）： 单品分摊的投放成本 = 总投放成本 × (该单品引流占比) ROI 计算公式 渠道 ROI = (渠道带来的净收入 − 渠道投放成本) / 渠道投放成本 单品 ROI = (单品净收入 − 单品分摊投放成本 − 产品成本) / 单品分摊投放成本 边际 ROI = Δ净收入 / Δ投放成本（判断”多投 1 万值不值”） 边际效用递减模型 投放额与转化不是线性关系：

• 前 20% 预算 → 带来 50% 转化（高效区）

• 中间 60% 预算 → 带来 40% 转化（递减区）

• 最后 20% 预算 → 仅带来 10% 转化（低效区） 模型：转化 = 总潜在转化 × (1 − e^(-k × 投放额/预算))

# -*- coding: utf-8 -*-
"""
config.py
渠道参数与转化基准值配置中心
"""
from typing import Dict
# ========== 渠道基础参数 ==========
CHANNEL_PARAMS = {
'short_video': {
'name': '短视频（抖音/小红书/快手）',
'cpm': 80.0, # 千次曝光成本（元）
'ctr': 0.025, # 点击率（2.5%）
'cvr': 0.015, # 转化率（1.5%）
'return_rate': 0.30, # 退货率（30%）
'brand_weight': 0.6, # 归因权重
'production_cost': 5000, # 内容制作成本（元/条）
'avg_views_per_post': 500000, # 平均单条播放量
},
'magazine': {
'name': '时尚杂志广告',
'cpm': 350.0, # 千次阅读成本
'ctr': 0.006, # 扫码/搜索点击率
'cvr': 0.05, # 杂志读者转化率高
'return_rate': 0.08, # 退货率低
'brand_weight': 0.3,
'production_cost': 20000, # 拍摄+版面费
'avg_readership': 200000, # 平均单期阅读量
},
}
# ========== 行业基准参数 ==========
INDUSTRY_BENCHMARKS = {
'avg_order_value': {
'apparel': 299.0, # 服饰类客单价
'accessory': 159.0, # 配饰类
'footwear': 399.0, # 鞋类
},
}

# ========== 边际效用模型参数 ==========
MARGINAL_PARAMS = {
'saturation_k': 0.00003, # 饱和系数（越小越容易饱和）
'efficiency_threshold': 0.15, # ROI < 15% 视为低效
'optimal_roi_range': (0.3, 0.8), # 最优 ROI 区间
}
# ========== 可视化配色 ==========
COLORS = {
'video': '#FF2442', # 小红书红
'magazine': '#333333', # 杂志黑
'positive': '#4CAF50', # 正 ROI
'negative': '#F44336', # 负 ROI
'neutral': '#607D8B', # 中性
'optimal': '#FF9800', # 最优区间
}
# ========== 数据模型文件 ==========
data_models = {
'bag': 599.0, # 包类
'logistics_cost': {
'apparel': 12.0, # 单件物流成本
'accessory': 8.0,
'footwear': 15.0,
'bag': 12.0,
},
'product_cost_ratio': {
'fast_fashion': 0.25, # 快时尚
'mid_market': 0.35, # 中端
'premium': 0.45, # 中高端
'luxury': 0.60, # 奢侈
},
}

重构后的内容： 数据模型层：渠道 / 单品 / 投放 / 核算结果

from dataclasses import dataclass, field
from typing import Dict, List, Optional
from enum import Enum
class ChannelType(Enum):
SHORT_VIDEO = "short_video"
MAGAZINE = "magazine"
class ProductCategory(Enum):
APPAREL = "apparel"
ACCESSORY = "accessory"
FOOTWEAR = "footwear"
BAG = "bag"
@dataclass
class Product:
product_id: str
name: str
category: ProductCategory
price: float # 售价
cost: float # 产品成本
logistics_cost: float # 单件物流
brand_tier: str = "mid_market" # 品牌定位
def profit_per_unit(self) -> float:
"""单件毛利"""
return self.price - self.cost - self.logistics_cost
def to_dict(self) -> Dict:
"""单品字典表示"""
return {
'单品ID': self.product_id,
'名称': self.name,
'品类': self.category.value,
'售价': self.price,
'成本': self.cost,
'物流': self.logistics_cost,
'单件毛利': round(self.profit_per_unit(), 2),
}

@dataclass class ChannelBudget: “”“渠道投放预算”“” channel: str budget: float production_count: int = 1 def total_cost(self, production_unit_cost: float = 0) -> float: “”“总投入 = 投放预算 + 制作成本”“” return self.budget + production_unit_cost * self.production_count @dataclass class FunnelResult: “”“单渠道漏斗结果”“” channel: str budget: float impressions: int = 0 clicks: int = 0 orders: int = 0 returns: int = 0 valid_orders: int = 0 ad_spend: float = 0.0 production_cost: float = 0.0 total_spend: float = 0.0 revenue: float = 0.0 net_profit: float = 0.0 roi: float = 0.0 def to_dict(self) -> Dict: “”“返回字典格式的数据”“” return { “渠道”: self.channel, “投放预算”: self.budget, “总投入”: round(self.total_spend, 2), “曝光量”: self.impressions, “点击量”: self.clicks, “下单量”: self.orders, “退货量”: self.returns, “有效成交”: self.valid_orders }

重构后的代码如下：
```python
class ProductRoiResult:
def __init__(self, product_id, product_name, channel_breakdown):
self.product_id = product_id
self.product_name = product_name
self.channel_breakdown = channel_breakdown
self.total_revenue = 0.0
self.total_cost = 0.0
self.total_profit = 0.0
self.roi = 0.0
self.orders = 0
self.rank = 0
@property
def to_dict(self):
return {
"排名": self.rank,
"单品": self.product_name,
"总收入": round(self.total_revenue, 2),
"总成本": round(self.total_cost, 2),
"净利润": round(self.total_profit, 2),
"ROI": f"{self.roi * 100:.1f}%",
"成交数": self.orders,
"渠道明细": self.channel_breakdown,
}
class OptimizationResult:
def __init__(self, current_total_roi, suggested_budget_shift, projected_roi_improvement):
self.current_total_roi = current_total_roi
self.suggested_budget_shift = suggested_budget_shift
self.projected_roi_improvement = projected_roi_improvement

在这段代码中，我们首先定义了一个名为ProductRoiResult的类，它包含了单品的ROI核算结果。这个类有一个构造函数，用于初始化产品ID、产品名称、渠道细分以及总收入、总成本、净利润和ROI等属性。此外，它还提供了一个名为to_dict的方法，该方法返回一个字典，其中包含了单品的排名、总收入、总成本、净利润、ROI、成交数、渠道明细等信息。 接下来，我们定义了一个名为OptimizationResult的类，它包含了预算优化建议。这个类有两个属性：current_total_roi表示当前的总ROI，suggested_budget_shift表示建议的预算调整，而projected_roi_improvement表示预计的ROI改进。 重构内容：

# 定义一个名为FunnelModel的类，用于表示流量漏斗模型
class FunnelModel:
def __init__(self, channel: ChannelType, params: Dict = None):
self.channel = channel
self.params = params or CHANNEL_PARAMS[channel.value]
def calculate(self, budget: float, product_price: float, production_cost: float = 0.0, production_count: int = 1):
"""
核心计算：给定预算，算出完整漏斗
Args:
budget (float): 投放预算（元）
product_price (float): 单品售价
production_cost (float, optional): 单条/单版制作成本
production_count (int, optional): 制作数量
Returns:
FunnelResult: 包含完整漏斗数据
"""
result = FunnelResult(
name=self.params["name"],
budget=budget,
)
return result

在这个重构版本中，我将funnel_model.py文件的内容进行了拆分和重构。首先，我定义了一个名为FunnelModel的类，它包含了两个主要的方法：__init__和calculate。__init__方法用于初始化模型，包括通道类型和参数。calculate方法用于根据给定的预算、产品价格、制作成本和制作数量来计算完整的漏斗数据。最后，我创建了一个新的名为FunnelResult的类，用于表示完整的漏斗数据。

计算结果

result = calculate_with_diminishing_returns( budget: float, product_price: float, production_cost: float = 0.0, production_count: int = 1, saturation_k: float = 0.00003 )

曝光量计算

result.impressions = int(budget / self.params[‘cpm’] * 1000)

点击量计算

result.clicks = int(result.impressions * self.params[‘ctr’])

下单量计算

result.orders = int(result.clicks * self.params[‘cvr’])

退货量计算

result.returns = int(result.orders * self.params[‘return_rate’])

有效成交计算

result.valid_orders = result.orders - result.returns

成本计算

result.ad_spend = budget result.production_cost = production_cost * production_count result.total_spend = budget + result.production_cost

收入计算

result.revenue = result.valid_orders * product_price

净利润计算

result.net_profit = result.revenue - result.total_spend

ROI计算

if result.total_spend > 0: result.roi = result.net_profit / result.total_spend return result def calculate_with_diminishing_returns( budget: float, product_price: float, production_cost: float = 0.0, production_count: int = 1, saturation_k: float = 0.00003 ):

计算结果

result = calculate_with_diminishing_returns( budget: float, product_price: float, production_cost: float = 0.0, production_count: int = 1, saturation_k: float = 0.00003 )

曝光量计算

result.impressions = int(budget / self.params[‘cpm’] * 1000)

点击量计算

result.clicks = int(result.impressions * self.params[‘ctr’])

下单量计算

result.orders = int(result.clicks * self.params[‘cvr’])

退货量计算

result.returns = int(result.orders * self.params[‘return_rate’])

有效成交计算

result.valid_orders = result.orders - result.returns

成本计算

result.ad_spend = budget result.production_cost = production_cost * production_count result.total_spend = budget + result.production_cost

收入计算

result.revenue = result.valid_orders * product_price

净利润计算

result.net_profit = result.revenue - result.total_spend

ROI计算

if result.total_spend > 0: result.roi = result.net_profit / result.total_spend return result

带边际效用递减的漏斗计算
模型：有效转化 = 理论转化 × (1 − e^(-k × 预算))
即：投得越多，每多 1 元的回报越低
result = self.calculate(budget, product_price, production_cost, production_count)
# 应用饱和函数修正有效成交
saturation_factor = 1 - np.exp(-saturation_k * budget)
theoretical_orders = result.valid_orders
result.valid_orders = int(theoretical_orders * saturation_factor)
result.orders = int(result.orders * saturation_factor)
result.returns = result.orders - result.valid_orders
# 重算收入利润
result.revenue = result.valid_orders * product_price
result.net_profit = result.revenue - result.total_spend
if result.total_spend > 0:
result.roi = result.net_profit / result.total_spend
return result
@staticmethod
def batch_calculate(
budgets: List[ChannelBudget],
product_price: float,
product_cost: float = 0.0,
) -> List[FunnelResult]:
# 批量计算多渠道漏斗
results = []
for cb in budgets:
model = FunnelModel(cb.channel)
params = CHANNEL_PARAMS[cb.channel.value]
# 计算有效转化
effective_conversion = model.calculate(product_price, product_cost, production_cost, production_count)
result = self.calculate(effective_conversion, product_price, product_cost, production_count)
# 应用饱和函数修正有效成交
saturation_factor = 1 - np.exp(-saturation_k * effective_conversion)
theoretical_orders = result.valid_orders
result.valid_orders = int(theoretical_orders * saturation_factor)
result.orders = int(result.orders * saturation_factor)
result.returns = result.orders - result.valid_orders
# 重算收入利润
result.revenue = result.valid_orders * product_price
result.net_profit = result.revenue - result.total_spend
if result.total_spend > 0:
result.roi = result.net_profit / result.total_spend
results.append(result)
return results

def time_decay(
orders: int,
channel_weights: Dict[str, float] = None,
) -> Dict[str, int]:
"""时间衰减归因：根据每个渠道的权重进行加权平均分配。"""
if channel_weights is None:
channel_weights = {
"short_video": 0.6,
"magazine": 0.3,
}
# 简化：按比例分配
total_weight = sum(channel_weights.values())
for ch, w in channel_weights.items():
results[ch] = int(orders * w / total_weight)
return results
@staticmethod
def linear_distribution(
orders: int,
channel_weights: Dict[str, float] = None,
) -> Dict[str, int]:
"""线性分配归因：根据每个渠道的权重进行直接分配。"""
if channel_weights is None:
channel_weights = {
"short_video": 0.5,
"magazine": 0.5,
}
# 简化：按比例分配
total_weight = sum(channel_weights.values())
for ch, w in channel_weights.items():
results[ch] = int(orders * w / total_weight)
return results

def time_decay(
orders: int,
channel_contact_days: Dict[str, int],
channel_weights: Dict[str, float] = None,
) -> Dict[str, float]:
"""
时间衰减归因：越近的触点权重越高
Args:
orders: 总订单数
channel_contact_days: {渠道: 接触距今天数}
channel_weights: 渠道基础权重
"""
if channel_weights is None:
channel_weights = {
'short_video': 0.6,
'magazine': 0.3,
}
# 计算衰减权重
decayed = {}
for ch, days in channel_contact_days.items():
base = channel_weights.get(ch, 0.1)
decay = 0.8 ** days # 每天衰减 20%
decayed[ch] = base * decay
total = sum(decayed.values())
results = {}
for ch, w in decayed.items():
results[ch] = orders * w / total
return results
@staticmethod
def distribute_revenue(
total_revenue: float,
channel_orders: Dict[str, int],
) -> Dict[str, float]:
"""按订单数分配收入到各渠道"""
total_orders = sum(channel_orders.values())
if total_orders == 0:
return {ch: 0.0 for ch in channel_orders}
return {
ch: total_revenue * orders / total_orders
}

for ch, orders in channel_orders.items():
video_budget = sum(order['video_budget'] for order in orders)
mag_budget = sum(order['mag_budget'] for order in orders)
# 跨渠道协同效应
cross_channel_synergy(video_budget, mag_budget)
@staticmethod
def cross_channel_synergy(
video_budget: float,
mag_budget: float,
):
"""
跨渠道协同效应
短视频 + 杂志同时投放时，转化率有协同提升
因为杂志建立信任 → 短视频促成冲动
"""
total = video_budget + mag_budget
if total == 0:
return 0.0
# 协同系数：两渠道预算越均衡，协同越强
video_ratio = video_budget / total
mag_ratio = mag_budget / total
# 均衡度 = 1 − |ratio − 0.5| × 2
balance = 1 - abs(video_ratio - 0.5) * 2
synergy_boost = 1 + balance * 0.25 # 最多提升 25%
return synergy_boost

根据提供的内容，以下是重构后的内容：

self.funnel_model = None
self.attribution = AttributionModel()
def calculate_channel_roi(
self,
channel_budget: ChannelBudget,
product: Product,
use_diminishing: bool = True,
):
"""计算单渠道 ROI"""
from config import CHANNEL_PARAMS, MARGINAL_PARAMS
model = FunnelModel(channel_budget.channel)
params = CHANNEL_PARAMS[channel_budget.channel.value]
prod_cost = params.get("production_cost", 0)
if use_diminishing:
k = MARGINAL_PARAMS["saturation_k"]
return model.calculate_with_diminishing_returns(
channel_budget.budget, product.price,
prod_cost, channel_budget.production_count, k
)
else:
return model.calculate(
channel_budget.budget, product.price,
prod_cost, channel_budget.production_count
)
def calculate_product_roi(
self,
product: Product,
channel_budgets: List[ChannelBudget],
):
"""核算单品在多渠道投放下的综合 ROI"""
channel_results = {}
# 核心逻辑：
# 1. 计算每个渠道的漏斗
# 2. 按渠道权重分配订单和收入
# 3. 汇总计算单品总 ROI
for channel in channel_budgets:
results = self.calculate_channel_roi(channel, product)
channel_results[channel.name] = results
return ProductRoiResult(channel_results)

这段代码定义了两个函数：calculate_channel_roi 和 calculate_product_roi。它们分别用于计算单品在多渠道投放下的ROI。在 calculate_channel_roi 函数中，我们首先导入必要的模块和类，然后根据参数设置模型、参数和产品成本。接下来，根据是否使用递减回报，我们调用不同的方法来计算ROI。最后，在 calculate_product_roi 函数中，我们遍历所有渠道，对每个渠道调用 calculate_channel_roi 函数，并将结果存储在一个字典中。最后，我们将这个字典转换为一个 ProductRoiResult 对象并返回。

total_orders = 0
total_revenue = 0.0
total_ad_spend = 0.0
total_production = 0.0
for cb in channel_budgets:
fr = self.calculate_channel_roi(cb, product)
channel_results[cb.channel.value] = {
'orders': fr.orders,
'valid_orders': fr.valid_orders,
'revenue': fr.revenue,
'ad_spend': fr.ad_spend,
'production_cost': fr.production_cost,
'roi': fr.roi,
'ctr': fr.clicks / fr.impressions if fr.impressions > 0 else 0,
'cvr': fr.orders / fr.clicks if fr.clicks > 0 else 0,
}
total_orders += fr.valid_orders
total_revenue += fr.revenue
total_ad_spend += fr.ad_spend
total_production += fr.production_cost
# 产品成本（按有效成交计算）
product_cost_total = product.cost * total_orders
logistics_total = product.logistics_cost * total_orders
total_cost = total_ad_spend + total_production + product_cost_total + logistics_total
net_profit = total_revenue - total_cost
roi = (total_revenue - total_ad_spend - total_production) / total_cost

内容重构如下：

if total_cost > 0:
return ProductRoiResult(
product_id=product.product_id,
product_name=product.name,
channel_breakdown=channel_results,
total_revenue=total_revenue,
total_cost=total_cost,
total_profit=net_profit,
roi=roi,
orders=total_orders,
)
def rank_products(
self,
products: List[Product],
channel_budgets: List[ChannelBudget],
):
"""
多单品 ROI 排名
"""
results = []
for p in products:
r = self.calculate_product_roi(p, channel_budgets)
results.append(r)
# 按 ROI 排序
results.sort(key=lambda x: x.roi, reverse=True)
# 标注排名
for i, r in enumerate(results):
r.rank = i + 1
return results

class MarginalAnalyzer:
"""
边际效用分析器
核心问题：
已经投了 X 万，再投 1 万，ROI 是升还是降？
"""
MARGINAL_PARAMS = {"saturation_k": 0.5} # 假设饱和点为 0.5
def __init__(self, product: Product):
"""
初始化方法
self.product = product
"""
self.product = product
self.saturation_k = MARGINAL_PARAMS[\"saturation_k\"]
def analyze(
self,
channel: str,
current_budget: float,
production_cost: float = 0.0,
step: float = 10000.0,
max_budget: float = 500000.0,
) -> List[Dict]:
"""
边际 ROI 分析
Returns:
每个预算水平下的边际 ROI
"""
from config import CHANNEL_PARAMS
channel_type = None
for ct, params in CHANNEL_PARAMS.items():
if params['name'] == channel or ct == channel:
channel_type = ct
break
if channel_type is None:
raise ValueError(f'未知渠道: {channel}')
# 计算当前预算下的预期收益
expected_revenue = self.product.calculate_expected_revenue()
# 计算实际收益和预期收益的差值
actual_revenue = self.product.calculate_actual_revenue()
marginal_roi = (actual_revenue - production_cost) / current_budget * 100
return [{'channel': channel, 'current_budget': current_budget, 'production_cost': production_cost, 'step': step, 'max_budget': max_budget, 'marginal_roi': marginal_roi}]