开场
某私募周四下午,团队为沪深300 ETF 期权准备了四个定价器口味——Black-Scholes 看涨、Black-Scholes 看跌、二叉树、蒙特卡洛。研究主管开了一次代码评审,发现生产代码里有一个 StrategyFactoryAbstract 抽象类、两个 AbstractPricerBuilder 子类、Confluence 上一张 60 行的依赖关系图,以及一个唯一目的就是包了一个 dict 的 SingletonRegistryProvider。这些名字每一个都是把 Java 模式逐字搬进 Python。功能形状大致没错,代码行数却是 Python 习惯写法的六倍。GoF 目录写于一门没有头等函数、没有运行时类对象、没有装饰器、没有鸭子类型的语言;Python 自 1991 年就具备这四样,所以大部分 GoF 模式在 Python 里会塌缩到十行鸭子类型代码。本课就是把目录用 Python 风格重写:五个模式完整讲,另外三个一段话讲清——恰好够用就行。
Strategy:三层深度,一条选型规则
Strategy 通过把会变的行为传进去,给算法做参数化。Python 里有三层深度。(a) 传一个普通函数:sorted(items, key=lambda x: x.price)——key 形参就是策略本身。(b) 传一个 Callable[..., T] 带类型的形参,当策略不是一行能写完但仍只有一个方法时。(c) 传一个满足 L1 Protocol 的对象,当策略在多个方法上是多态的——例如 Pricer 既要 price 又要 delta。选型规则:无状态 → 函数;带参数 + 单方法 → Callable;多态 + 多方法 → Protocol。
# (a) plain function as strategy
sorted(quotes, key=lambda q: q.price)
# (b) Callable-typed parameter
def run_backtest(signal: Callable[[pd.DataFrame], pd.Series]) -> Stats: ...
# (c) Protocol-typed parameter (multiple methods)
def run_session(pricer: Pricer) -> None: result = pricer.price(s); risk = pricer.delta(s)
一个能接受三种形式的 run_pricing(strategy, *args) 驱动器只有两行:if callable(strategy): return strategy(*args),否则 return strategy.price(*args)。没有 StrategyBase,没有 StrategyContext,没有抽象工厂。
Adapter / Facade:一个接口,多家供应商
国内 量化团队在 2026 年通常会把行情数据适配器按供应商分开(Tushare / 万得 / 米筐 / Choice),统一到一个 facade 后面,因为中途换供应商是常态。每个内部调用方都用同一个形状:get_bars(ticker, start, end) -> pd.DataFrame,列模式固定。在这个函数背后是每家供应商各一个 adapter,把统一调用翻译成各家 SDK 的具体形状。
class MarketDataFacade(Protocol):
def get_bars(self, ticker: str, start: str, end: str) -> pd.DataFrame: ...
class StubbedAdapter:
def get_bars(self, ticker: str, start: str, end: str) -> pd.DataFrame:
# vendor-specific call goes here
...
def run_strategy(md: MarketDataFacade, ticker: str) -> None: bars = md.get_bars(ticker, '2024-01-01', '2024-01-31'); ...
规则是明确的:调用方依赖 facade 接口;adapter 依赖供应商 SDK;调用方绝不直接 import 供应商 SDK。一个 Tushare 风格的具体 adapter 在内部把 ticker 翻成 ts_code='510300.SH',把 start='2024-01-01' 翻成 start_date='20240101' 再发出去,调用方完全无感。L1 的 Protocol 是接口;具体 adapter 是结构上满足它的实现——无需继承。
Factory + Registry:一个 dict 加一个装饰器
Java 模式(AbstractPricerFactory → BlackScholesPricerFactory 等等)等价于一个模块级 dict 加一个注册装饰器。具体定价器在 import 时自注册:
_PRICERS: dict[str, type] = {}
def register(name: str):
def deco(cls):
_PRICERS[name] = cls
return cls
return deco
def make_pricer(name: str, **kwargs):
return _PRICERS[name](**kwargs)
@register('bs-call')
class BlackScholesCallPricer:
def __init__(self, s: float, k: float, r: float, sigma: float, t: float) -> None: ...
def price(self) -> float: ...
510300.SH 的定价器注册表包含 bs-call、bs-put、binomial-call、monte-carlo-call,统一用 S_0 = 4.20 元、K = 4.30 元、r = 0.025、sigma = 0.22、T = 30/252 这一组参数。make_pricer('bs-call', s=4.20, k=4.30, r=0.025, sigma=0.22, t=30/252).price() 返回看涨期权价。整个工厂只有六行加一个装饰器——没有 AbstractPricerFactory,因为模块级状态本来就是 Python 风格的单例容器。
这里要留意 import 的副作用。@register 装饰器只在定价器模块被 import 时才会触发;注册表为空通常意味着忘记 import。规范的解决方式是用一个 __init__.py 把每个具体定价器都 import 进来,或者写一个 pkgutil.iter_modules 遍历,把 pricers/ 包里的全部模块都 import 一遍。当第三方包要注册自家定价器时,Python 的入口点机制(pyproject.toml + importlib.metadata.entry_points)是注册表模式的标准延伸——在 pyproject.toml 写一行声明,定价器就在 install 时被注册。
Observer / 发布订阅:WeakSet 防漏
主体在不知道谁在订阅的情况下通知一组会变的订阅者。Python 风格使用 weakref.WeakSet 防止「忘记 unsubscribe」导致的泄漏——当订阅者对象的强引用全部消失,WeakSet 会悄悄把它丢掉:
import weakref
class EventBus:
def __init__(self) -> None: self._subs: weakref.WeakSet = weakref.WeakSet()
def subscribe(self, cb) -> None: self._subs.add(cb)
def dispatch(self, evt) -> None:
for cb in list(self._subs):
cb(evt)
bus = EventBus(); bus.subscribe(strategy_sub); bus.subscribe(logger_sub); bus.dispatch(quote)
TickBus.dispatch(quote) 把 510300.SH 的 PriceQuote 事件扇出到一个跑沪深300 动量策略的 StrategySubscriber、一个写 stdout 的 LoggerSubscriber 以及一个用元统计累计盈亏的 PnLTrackerSubscriber。推 vs. 拉的选择:推(发布者内联调用订阅者)更简单,但有一个慢订阅者就能把发布者堵住。拉(订阅者从队列里取)适合订阅者慢或远端的场景,这也是通往第三课生产者-消费者模式的自然桥梁——那时你就把 WeakSet 换成 queue.Queue。
依赖注入:把协作者作为参数传进来
Java 模式(运行时通过反射布线构造器的 DI 容器)在 Python 里被「把协作者作为参数传进来」替代。一个需要 Pricer 和 DataSource 的类在 __init__(self, pricer, data) 里接它们,调用方决定传什么——测试传 InMemoryDataSource 适配器,生产 main 传 TushareAdapter,notebook 想传啥传啥。没有框架;模式就是不去抓全局这件事的纪律。测试替换的好处是免费的:没有 monkeypatch、没有 unittest.mock、没有 patching。(参考 3.1.3 L3 的对照——monkeypatch 是「你改不了调用方时」的对的工具,而 DI 纪律的目的正是让你能改调用方。)补一句:FastAPI 的 Depends 是大多数 Python 团队会碰到的唯一一个框架式 DI 接口,3.6.5 会讲。
三个模式,各一段
Singleton。Python 风格的单例就是模块级对象;import config; config.broker_url = '...' 之所以可行,是因为模块命名空间本身就是首次 import 时被填一次的单一共享作用域。不要写 __new__ 的把戏类。唯一例外是必须继承的场景,可用 Borg 模式——class Borg: _shared_state = {}; def __init__(self): self.__dict__ = self._shared_state——共享状态但不共享 identity;否则用模块。Visitor。用 functools.singledispatch 替代,按第一个参数的类型派发:@singledispatch def render(node): ...; @render.register(IfNode) def _(node): ...; @render.register(LoopNode) def _(node): ...。Decorator(GoF 含义,不是 @deco 语法)。@deco 语法本身加上 3.1.2 L3 的 functools.wraps 与 lru_cache 已经覆盖了 GoF Decorator 模式的大部分场景;要在方法外侧加一层职责的协议装饰子集,写一个通过 __getattr__ 委托的包装类即可。
Python 风格替换速查表
| classical pattern | Pythonic replacement | key language feature | one-line example from this lesson |
|---|---|---|---|
Strategy | callable or Protocol | first-class functions + structural typing | sorted(quotes, key=lambda q: q.price) |
AbstractFactory | registry dict | module-level state + class decorators | make_pricer('bs-call', s=..., k=..., r=..., sigma=..., t=...) |
Singleton | module-level state | module namespace is a single shared scope | import config; config.broker_url = ... |
Visitor | functools.singledispatch | type-based single dispatch | @render.register(IfNode) def _(node): ... |
四行口诀:策略 -> 可调用对象或 Protocol;抽象工厂 -> 注册字典;单例 -> 模块级状态;访问者 -> functools.singledispatch。一句收尾的告诫:模式是解法的形状,不是字面上的类层级。如果你团队的 Python 生产代码里有 StrategyFactoryAbstract,你搬过来的是名词而非想法。GoF 一书仍留在阅读清单上,因为模式的分类法有价值;其实现则大多是缺失语言特性的 workaround,Python 翻译会把它压扁。
练习
Exercise
搭一个迷你定价器注册表。(a) 在模块作用域定义 _REGISTRY: dict[str, type] = {}。(b) 写一个 register(name: str) 装饰器,它把被装饰的类按 name 插入 _REGISTRY 并原样返回该类。(c) 把 @register("square") 套到 class SquarePricer 上,它的 __init__(self, x: float) 保存 x,price(self) -> float 返回 self.x ** 2。(d) 写 make_pricer(name: str, **kwargs),它返回 _REGISTRY[name](**kwargs)。(e) 验证 make_pricer("square", x=4).price() == 16.0。(f) 用一句话说明:如果 SquarePricer 定义在另一个模块里,必须发生什么 import 副作用,"square" 才会出现在 _REGISTRY 里。
提示
register 返回 deco;deco 接一个类,改 _REGISTRY,原样返回该类,这样 class SquarePricer: 这一名字仍指向原类。提示
class SquarePricer: 这条语句时才发生——这要求模块至少被 import 过一次。通往下一课的桥
你现在能在没有 StrategyBase 的情况下表达 Strategy,能把行情数据走一个供应商无关的 facade,能用六行工厂注册定价器,能给观察者扇出做防泄漏的订阅,也知道在 Java 用 Visitor 的位置改用 singledispatch。观察者模式恰好暗示了下一课的位置。当订阅者慢时,推式分发会让一个差消费者把所有其他消费者都堵住;办法是把发布者的内联调用换成一个队列,让消费者按自己的节奏排空。队列是 queue.Queue,纪律是生产者-消费者模式,让它线程安全所需的锁与信号量原语,正是下一课要覆盖的内容——还有它们的 asyncio 对应版本。
阅读清单:《Fluent Python》第 2 版 中译 第 10 章 (函数作为一等对象) 与 第 11 章 (函数装饰器与闭包) 作为 Strategy / Decorator 的基础, 第 26 章 (类元编程) 的注册表小节; 《设计模式:可复用面向对象软件的基础》中译 (GoF 原书) 作为对照阅读; Peter Norvig 的 Design Patterns in Dynamic Languages 1996 年演讲讲义 (英文; 国内有中文翻译) 是本课论点的源头; PEP 443 (singledispatch) 的中文社区翻译。