以下内容围绕策略模式在 Python 的作用、结构与实践展开，并包含一个面向结算域（Checkout）的进阶 FastAPI 服务端示例，通过策略模式将促销与税费规则解耦，支持渠道与会员等级组合、运行时策略链选择、A/B 测试与灰度发布。
一、策略模式在 Python 的作用
- 目标：将算法（业务规则）封装为可互换的策略对象，通过统一接口在运行时选择和组合，避免大量 if-else/switch。
- 价值：
- 解耦：调用方不依赖具体促销/税费规则，实现模块化和可插拔。
- 扩展：新增策略不影响既有代码，开放封闭原则（OCP）。
- 测试友好：策略是纯粹的小单元，可独立编写和测试。
- 部署灵活：可根据配置中心策略键动态启用/禁用与组合，适配 A/B 和灰度。
二、结构（关键组件与角色）
- Strategy 接口：定义统一行为（如 apply(context)），可包含适用性判断 is_applicable(context)。
- ConcreteStrategy：具体策略实现（如百分比折扣、满减、阶梯价、包邮、税费）。
- Context（业务上下文）：承载结算输入与中间结果（价格、税费、运费、用户标签、渠道、站点等）。
- StrategyRegistry：策略注册表，按“类型”生成策略实例，支持根据配置构建策略链。
- ConfigService（外部配置中心代理）：从配置下发策略键与参数，支持 A/B/灰度。
- CheckoutService（应用上下文）：根据请求信息和配置构造策略链并执行，返回价格拆解。
三、常见使用场景
- 电商结算：折扣、满减、阶梯价、包邮、税费、多币种换算、复杂排他/叠加规则。
- 订阅/会员：分层价、按渠道/地域差异定价、合规税费。
- 营销试验：A/B 测试、策略灰度发布、按用户标签/站点切换。
四、优缺点
- 优点：
- 高内聚低耦合：策略可被独立替换与组合。
- 运行时可配置：面向策略键的动态加载。
- 易测试：每个策略单测简单明确。
- 缺点：
- 策略爆炸：策略数量多时需要良好治理（命名、注册、可见性）。
- 顺序与相互影响需定义：促销、税、运费的执行顺序影响结果，需在架构层明确排序与约束。
- 配置复杂度：配置中心需要良好的参数校验与回滚机制。
五、Python FastAPI 服务端示例（结算域：促销与税费解耦、策略链组合）
说明：
- 设计 PriceStrategy 接口，具体实现包含：百分比折扣、满减、阶梯价、包邮；附加 VAT 税费策略展示税费解耦。
- 支持按渠道与会员等级组合。
- CheckoutService 根据站点、用户标签与配置中心的策略键，在运行时选择并组合策略链。
- 结构单元测试友好，含策略扩展模板。
代码示例（可直接运行的结构化示例，省略持久化与真实配置中心接入，以内存模拟配置下发）：
from __future__ import annotations
from typing import List, Dict, Any, Optional, Callable
from decimal import Decimal, ROUND_HALF_UP, getcontext
from dataclasses import dataclass, replace
from enum import Enum
from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel, Field, condecimal, conint, validator
# -----------------------------
# Money & enums
# -----------------------------
getcontext().prec = 28 # sufficient precision
def money(v: Decimal) -> Decimal:
return v.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
class Channel(str, Enum):
web = "web"
mobile = "mobile"
partner = "partner"
class MemberLevel(str, Enum):
standard = "standard"
silver = "silver"
gold = "gold"
# -----------------------------
# Pydantic models (request/response)
# -----------------------------
class LineItem(BaseModel):
sku: str = Field(..., min_length=1)
qty: conint(gt=0) = 1
unit_price: condecimal(gt=0, max_digits=12, decimal_places=2)
class CheckoutRequest(BaseModel):
site: str = Field(..., min_length=1)
channel: Channel
member_level: MemberLevel
user_tags: List[str] = []
currency: str = "CNY"
items: List[LineItem]
base_shipping_fee: condecimal(ge=0, max_digits=12, decimal_places=2) = Decimal("0.00")
ab_bucket: Optional[str] = None
@validator("items")
def non_empty_items(cls, v):
if not v:
raise ValueError("items cannot be empty")
return v
class StrategyApplication(BaseModel):
key: str
description: str
delta: condecimal(max_digits=12, decimal_places=2)
meta: Dict[str, Any] = {}
class CheckoutResponse(BaseModel):
currency: str
subtotal: condecimal(max_digits=12, decimal_places=2)
discount_total: condecimal(max_digits=12, decimal_places=2)
shipping_fee: condecimal(max_digits=12, decimal_places=2)
tax_total: condecimal(max_digits=12, decimal_places=2)
final_total: condecimal(max_digits=12, decimal_places=2)
applied: List[StrategyApplication]
# -----------------------------
# Runtime PricingContext
# -----------------------------
@dataclass(frozen=True)
class PricingContext:
site: str
channel: Channel
member_level: MemberLevel
user_tags: List[str]
currency: str
subtotal: Decimal
total_qty: int
discount_total: Decimal
shipping_fee: Decimal
tax_total: Decimal
final_total: Decimal
applied: List[StrategyApplication]
def with_updates(
self,
*,
discount_delta: Decimal = Decimal("0.00"),
shipping_delta: Decimal = Decimal("0.00"),
tax_delta: Decimal = Decimal("0.00"),
application: Optional[StrategyApplication] = None,
) -> "PricingContext":
new_discount = money(self.discount_total + discount_delta)
new_shipping = money(self.shipping_fee + shipping_delta)
new_tax = money(self.tax_total + tax_delta)
new_final = money(self.subtotal - new_discount + new_shipping + new_tax)
new_applied = list(self.applied)
if application:
new_applied.append(application)
return replace(
self,
discount_total=new_discount,
shipping_fee=new_shipping,
tax_total=new_tax,
final_total=new_final,
applied=new_applied,
)
# -----------------------------
# Strategy interface & registry
# -----------------------------
class PriceStrategy:
key: str
description: str
order: int # lower first; promotions < shipping < tax
def is_applicable(self, ctx: PricingContext) -> bool:
return True
def apply(self, ctx: PricingContext) -> PricingContext:
raise NotImplementedError
# Strategy spec from ConfigService
class StrategySpec(BaseModel):
type: str # e.g., "percentage_discount"
key: str
params: Dict[str, Any] = {}
order: int = 100 # default
StrategyFactory = Callable[[StrategySpec], PriceStrategy]
class StrategyRegistry:
def __init__(self):
self._builders: Dict[str, StrategyFactory] = {}
def register(self, type_name: str, builder: StrategyFactory):
self._builders[type_name] = builder
def build(self, spec: StrategySpec) -> PriceStrategy:
if spec.type not in self._builders:
raise KeyError(f"Unknown strategy type: {spec.type}")
strategy = self._builders[spec.type](spec)
# allow spec to override default order
strategy.order = spec.order
return strategy
# -----------------------------
# Concrete strategies
# -----------------------------
class PercentageDiscountStrategy(PriceStrategy):
def __init__(self, key: str, percent: Decimal, allowed_channels: List[Channel], allowed_levels: List[MemberLevel]):
self.key = key
self.description = f"{percent}% off"
self.percent = Decimal(percent)
self.allowed_channels = set(allowed_channels)
self.allowed_levels = set(allowed_levels)
self.order = 10
def is_applicable(self, ctx: PricingContext) -> bool:
return ctx.channel in self.allowed_channels and ctx.member_level in self.allowed_levels
def apply(self, ctx: PricingContext) -> PricingContext:
discount = money(ctx.subtotal * self.percent / Decimal("100"))
if discount <= Decimal("0.00"):
return ctx
return ctx.with_updates(
discount_delta=discount,
application=StrategyApplication(
key=self.key, description=self.description, delta=-discount, meta={"percent": str(self.percent)}
),
)
class FullReductionStrategy(PriceStrategy):
def __init__(self, key: str, rules: List[Dict[str, Any]], allowed_channels: List[Channel]):
"""
rules: [{threshold: 300, reduce: 40}, {threshold: 600, reduce: 90}]
pick the best rule for subtotal
"""
self.key = key
self.description = "Full reduction"
self.rules = sorted(rules, key=lambda r: Decimal(str(r["threshold"])))
self.allowed_channels = set(allowed_channels)
self.order = 20
def is_applicable(self, ctx: PricingContext) -> bool:
return ctx.channel in self.allowed_channels
def apply(self, ctx: PricingContext) -> PricingContext:
applicable = [r for r in self.rules if ctx.subtotal >= Decimal(str(r["threshold"]))]
if not applicable:
return ctx
best = max(applicable, key=lambda r: Decimal(str(r["reduce"])))
reduce_amt = money(Decimal(str(best["reduce"])))
return ctx.with_updates(
discount_delta=reduce_amt,
application=StrategyApplication(
key=self.key, description=f"满减: 满{best['threshold']}减{best['reduce']}", delta=-reduce_amt, meta=best
),
)
class TieredPricingStrategy(PriceStrategy):
def __init__(self, key: str, tiers: List[Dict[str, Any]], allowed_levels: List[MemberLevel]):
"""
tiers: [{min_qty: 3, percent: 5}, {min_qty: 5, percent: 10}]
applies percent discount based on total_qty
"""
self.key = key
self.description = "Tiered pricing by qty"
self.tiers = sorted(tiers, key=lambda t: int(t["min_qty"]))
self.allowed_levels = set(allowed_levels)
self.order = 15
def is_applicable(self, ctx: PricingContext) -> bool:
return ctx.member_level in self.allowed_levels
def apply(self, ctx: PricingContext) -> PricingContext:
applicable = [t for t in self.tiers if ctx.total_qty >= int(t["min_qty"])]
if not applicable:
return ctx
best = max(applicable, key=lambda t: Decimal(str(t["percent"])))
percent = Decimal(str(best["percent"]))
discount = money(ctx.subtotal * percent / Decimal("100"))
if discount <= Decimal("0.00"):
return ctx
return ctx.with_updates(
discount_delta=discount,
application=StrategyApplication(
key=self.key, description=f"阶梯价: 满{best['min_qty']}件打{percent}%", delta=-discount, meta=best
),
)
class FreeShippingStrategy(PriceStrategy):
def __init__(self, key: str, threshold: Optional[Decimal], allowed_channels: List[Channel], allowed_levels: List[MemberLevel]):
self.key = key
self.description = "Free shipping"
self.threshold = Decimal(str(threshold)) if threshold is not None else None
self.allowed_channels = set(allowed_channels)
self.allowed_levels = set(allowed_levels)
self.order = 30 # after promotions but before tax
def is_applicable(self, ctx: PricingContext) -> bool:
if ctx.channel not in self.allowed_channels or ctx.member_level not in self.allowed_levels:
return False
if self.threshold is None:
return True
# Apply free shipping if subtotal after discounts >= threshold
subtotal_after_discount = money(ctx.subtotal - ctx.discount_total)
return subtotal_after_discount >= self.threshold
def apply(self, ctx: PricingContext) -> PricingContext:
if ctx.shipping_fee <= Decimal("0.00"):
return ctx
waive = ctx.shipping_fee
return ctx.with_updates(
shipping_delta=-waive,
application=StrategyApplication(
key=self.key, description="包邮", delta=-waive, meta={"threshold": str(self.threshold) if self.threshold else None}
),
)
class VATTaxStrategy(PriceStrategy):
def __init__(self, key: str, vat_percent: Decimal):
self.key = key
self.description = "VAT tax"
self.vat_percent = Decimal(vat_percent)
self.order = 40 # taxes last
def is_applicable(self, ctx: PricingContext) -> bool:
return True
def apply(self, ctx: PricingContext) -> PricingContext:
taxable_base = money(ctx.subtotal - ctx.discount_total + ctx.shipping_fee)
tax = money(taxable_base * self.vat_percent / Decimal("100"))
if tax <= Decimal("0.00"):
return ctx
return ctx.with_updates(
tax_delta=tax,
application=StrategyApplication(
key=self.key, description=f"VAT {self.vat_percent}%", delta=tax, meta={"base": str(taxable_base)}
),
)
# -----------------------------
# Config service (simulated)
# -----------------------------
class ConfigService:
"""
Simulates strategy specs from config center.
In real system, this could read from a KV store / feature flag service.
"""
def get_strategy_specs(self, site: str, channel: Channel, member_level: MemberLevel, user_tags: List[str], ab_bucket: Optional[str]) -> List[StrategySpec]:
# Simplified rules for demo; would be data-driven in production
specs: List[StrategySpec] = []
# Example A/B buckets
bucket = ab_bucket or "A"
# Percentage discount: mobile + gold users get 10% in bucket A, 12% in bucket B
percent = "12" if bucket == "B" else "10"
specs.append(StrategySpec(
type="percentage_discount",
key=f"percent-{percent}-mobile-gold",
params={"percent": percent, "allowed_channels": ["mobile"], "allowed_levels": ["gold"]},
order=10,
))
# Tiered pricing for silver/gold
specs.append(StrategySpec(
type="tiered_pricing",
key="tier-qty",
params={"tiers": [{"min_qty": 3, "percent": 5}, {"min_qty": 5, "percent": 10}], "allowed_levels": ["silver", "gold"]},
order=15,
))
# Full reduction on web + mobile
specs.append(StrategySpec(
type="full_reduction",
key="full-300-40",
params={"rules": [{"threshold": 300, "reduce": 40}, {"threshold": 600, "reduce": 90}], "allowed_channels": ["web", "mobile"]},
order=20,
))
# Free shipping over 200 for web/mobile, gold only
specs.append(StrategySpec(
type="free_shipping",
key="free-ship-200",
params={"threshold": "200", "allowed_channels": ["web", "mobile"], "allowed_levels": ["gold"]},
order=30,
))
# VAT tax 7%
specs.append(StrategySpec(
type="vat_tax",
key="vat-7",
params={"vat_percent": "7"},
order=40,
))
return specs
# -----------------------------
# Registry DI setup
# -----------------------------
def create_registry() -> StrategyRegistry:
reg = StrategyRegistry()
reg.register("percentage_discount", lambda spec: PercentageDiscountStrategy(
key=spec.key,
percent=Decimal(str(spec.params["percent"])),
allowed_channels=[Channel(c) for c in spec.params.get("allowed_channels", [])],
allowed_levels=[MemberLevel(l) for l in spec.params.get("allowed_levels", [])],
))
reg.register("full_reduction", lambda spec: FullReductionStrategy(
key=spec.key,
rules=spec.params["rules"],
allowed_channels=[Channel(c) for c in spec.params.get("allowed_channels", [])],
))
reg.register("tiered_pricing", lambda spec: TieredPricingStrategy(
key=spec.key,
tiers=spec.params["tiers"],
allowed_levels=[MemberLevel(l) for l in spec.params.get("allowed_levels", [])],
))
reg.register("free_shipping", lambda spec: FreeShippingStrategy(
key=spec.key,
threshold=Decimal(str(spec.params.get("threshold"))) if "threshold" in spec.params else None,
allowed_channels=[Channel(c) for c in spec.params.get("allowed_channels", [])],
allowed_levels=[MemberLevel(l) for l in spec.params.get("allowed_levels", [])],
))
reg.register("vat_tax", lambda spec: VATTaxStrategy(
key=spec.key,
vat_percent=Decimal(str(spec.params["vat_percent"])),
))
return reg
# DI providers for FastAPI
def get_config_service() -> ConfigService:
return ConfigService()
def get_strategy_registry() -> StrategyRegistry:
return create_registry()
# -----------------------------
# Pipeline executor (test-friendly)
# -----------------------------
def execute_pricing_pipeline(
ctx: PricingContext,
specs: List[StrategySpec],
registry: StrategyRegistry,
) -> PricingContext:
strategies: List[PriceStrategy] = []
for spec in specs:
st = registry.build(spec)
strategies.append(st)
# sort by order; if same order, keep config order
strategies.sort(key=lambda s: s.order)
current = ctx
for s in strategies:
if s.is_applicable(current):
current = s.apply(current)
return current
# -----------------------------
# FastAPI app & handler
# -----------------------------
app = FastAPI(title="Checkout Pricing Service")
@app.post("/checkout/price", response_model=CheckoutResponse)
def checkout_price(
req: CheckoutRequest,
cfg: ConfigService = Depends(get_config_service),
reg: StrategyRegistry = Depends(get_strategy_registry),
):
subtotal = sum((Decimal(item.unit_price) * item.qty for item in req.items), start=Decimal("0.00"))
subtotal = money(subtotal)
total_qty = sum(item.qty for item in req.items)
initial = PricingContext(
site=req.site,
channel=req.channel,
member_level=req.member_level,
user_tags=req.user_tags,
currency=req.currency,
subtotal=subtotal,
total_qty=total_qty,
discount_total=Decimal("0.00"),
shipping_fee=money(Decimal(req.base_shipping_fee)),
tax_total=Decimal("0.00"),
final_total=money(subtotal + Decimal(req.base_shipping_fee)),
applied=[],
)
try:
specs = cfg.get_strategy_specs(req.site, req.channel, req.member_level, req.user_tags, req.ab_bucket)
final_ctx = execute_pricing_pipeline(initial, specs, reg)
except KeyError as e:
raise HTTPException(status_code=400, detail=str(e))
return CheckoutResponse(
currency=final_ctx.currency,
subtotal=money(final_ctx.subtotal),
discount_total=money(final_ctx.discount_total),
shipping_fee=money(final_ctx.shipping_fee),
tax_total=money(final_ctx.tax_total),
final_total=money(final_ctx.final_total),
applied=final_ctx.applied,
)
# -----------------------------
# Unit test hints (not executed here)
# -----------------------------
"""
Example test for strategies:
def test_full_reduction():
ctx = PricingContext(
site="cn", channel=Channel.web, member_level=MemberLevel.standard, user_tags=[],
currency="CNY", subtotal=Decimal("650.00"), total_qty=2,
discount_total=Decimal("0.00"), shipping_fee=Decimal("20.00"),
tax_total=Decimal("0.00"), final_total=Decimal("670.00"), applied=[]
)
spec = StrategySpec(
type="full_reduction", key="full-300-40",
params={"rules":[{"threshold":300,"reduce":40},{"threshold":600,"reduce":90}], "allowed_channels":["web"]},
order=20
)
reg = create_registry()
st = reg.build(spec)
assert st.is_applicable(ctx)
new_ctx = st.apply(ctx)
assert new_ctx.discount_total == Decimal("90.00")
assert any(a.key=="full-300-40" for a in new_ctx.applied)
# Integration test:
def test_pipeline_order():
reg = create_registry()
cfg = ConfigService()
# Build context and pipeline as in endpoint, assert final_total and applied order.
"""
# -----------------------------
# Strategy extension template
# -----------------------------
class NewPriceStrategyTemplate(PriceStrategy):
"""
Template for introducing a new strategy.
1) Define key, description, order.
2) Implement is_applicable(ctx) with channel/level/tags/geo rules if needed.
3) Implement apply(ctx) returning a NEW context via ctx.with_updates(...)
4) Register builder in StrategyRegistry.
"""
def __init__(self, key: str, params: Dict[str, Any]):
self.key = key
self.description = "New Strategy"
self.order = params.get("order", 25)
self.params = params
def is_applicable(self, ctx: PricingContext) -> bool:
# Example: only on partner channel and silver+
allowed_channels = set([Channel.partner])
return ctx.channel in allowed_channels and ctx.member_level in {MemberLevel.silver, MemberLevel.gold}
def apply(self, ctx: PricingContext) -> PricingContext:
# Example: fixed discount of 15
discount = money(Decimal("15.00"))
return ctx.with_updates(
discount_delta=discount,
application=StrategyApplication(key=self.key, description=self.description, delta=-discount, meta=self.params),
)
# To enable the template:
# reg.register("new_strategy_type", lambda spec: NewPriceStrategyTemplate(key=spec.key, params=spec.params))
设计要点与落地建议：
- 排序约定：促销类（折扣/满减/阶梯价）应先执行，再处理运费（包邮等），最后计算税费。通过策略的 order 字段定义。
- 可观察性：在 PricingContext.applied 保留策略应用日志，便于审计与问题定位。
- 配置治理：
- 为每类策略定义严格的参数模式（Schema）与校验，避免错误参数导致价格异常。
- 给策略键命名规范（type-params-channel-level），支持灰度回滚。
- 叠加与排他：
- 若需要复杂排他（如满减与百分比折扣不可叠加），可在 StrategyRegistry 增加“冲突检测”，或在 apply 中查询已应用的策略并决定跳过。
- 多租户与渠道隔离：
- 根据 site/channel/member_level/user_tags 过滤 is_applicable，避免策略越权。
- 钱包精度与货币：
- 全链路使用 Decimal 并统一量化到 0.01，必要时引入货币与税区（Geo）的抽象分层。
