企业官网建设流程全解析

因子计算与行情数据的工程化解耦：让因子不再被数据源绑架

一、为什么「拉数」和「算因子」必须分开

很多量化新手的第一版因子代码长这样：在脚本顶部写一个requests.get拉行情，紧接着就在同一段里算动量、算波动、算 z-score，最后把结果塞进回测。短期看没问题，但一旦规模上来，三个坑会同时爆发：

• 因子逻辑没法单测。你想验证「20 日动量公式对不对」，却必须先能联网、必须有证书、必须等接口返回——一个纯数学函数被数据源绑架了。
• 换源要改一堆地方。今天用这套接口，明天想加一个备用源对账，结果发现因子代码里到处硬编码了域名，改到怀疑人生。
• 限频和缓存无从下手。拉数和算因子耦合在一起，你要么每次重算都重新打接口，要么把缓存逻辑揉进因子函数，越写越脏。

社区里反复被推荐的「三层数据治理架构」——接入层统一封装、缓存层 LRU+TTL、验证层每日黄金集比对——其底层前提就是一句话：数据获取与因子计算必须解耦。本篇就用零 SDK 纯 HTTP 接口，演示一套最小可用的解耦骨架：一个数据源接口、一组纯函数因子、一个因子引擎，以及一层不侵入因子代码的缓存。

注：本文代码均为零 SDK 纯 HTTP GET 接入行情 API，先本地跑通再成文。演示证书返回固定样本，真实证书返回真实数据；本文重点是解耦架构，数据以真实证书为准。

二、本文你将得到什么

• 一个MarketDataSource抽象：把「去哪拿数据」和「怎么算因子」彻底隔离。
• 一组与数据源无关的纯函数因子：动量、已实现波动率、均值回归 z-score。
• 一个FactorEngine：喂进任意数据源，吐出同一套因子，代码零改动。
• 一层CachedDataSource：把缓存、重试、熔断都变成「装饰器」，不污染因子逻辑。
• 一个关键认知：因子逻辑可以完全脱离网络与证书做单测。

三、解耦骨架（已本地自验）

下面这段就是本篇可运行的核心。注意看：momentum / realized_vol / mean_reversion_z这三个因子函数，签名里只有bars，它们从来不知道数据来自接口还是来自本地合成——这正是解耦的目的。

# -*- coding: utf-8 -*-importmathimportrequests BASE="https://api.momaapi.com"LICENCE="TEST-API-TOKEN-MOMA-836089C22111"# 演示证书，请换成你自己的正式证书classBar:__slots__=("code","date","open","high","low","close","volume","amount")def__init__(self,code,date,o,h,l,c,v,a):self.code=code;self.date=date self.open=o;self.high=h;self.low=l;self.close=c self.volume=v;self.amount=aclassMarketDataSource:defget_daily_bars(self,code:str,n:int)->list:raiseNotImplementedErrorclassMomaDataSource(MarketDataSource):"""行情 API 实现：纯 HTTP GET，零 SDK。"""defget_daily_bars(self,code:str,n:int)->list:url=f"{BASE}/hsstock/history/{code}/d/n/{LICENCE}"rows=requests.get(url,timeout=20).json()return[Bar(code,r["t"],r["o"],r["h"],r["l"],r["c"],r["v"],r["a"])forrinrows[-n:]]classStubDataSource(MarketDataSource):"""合成数据源：确定性 sin+噪声，专供因子逻辑单测，不依赖网络。"""def__init__(self,code="STUB"):self.code=code p,self.prices=100.0,[]foriinrange(120):p=p*(1+0.01*math.sin(i/6.0)+0.003*math.cos(i/3.0))self.prices.append(round(p,4))defget_daily_bars(self,code:str,n:int)->list:px=self.prices[-n:]bars=[]fori,cinenumerate(px):o=px[i-1]ifi>0elsec bars.append(Bar(self.code,f"D{i:03d}",o,max(o,c),min(o,c),c,1_000_000,1.0e8))returnbarsdefmomentum(bars:list,n:int)->float:iflen(bars)<n+1:returnfloat("nan")returnbars[-1].close/bars[-1-n].close-1.0defrealized_vol(bars:list,n:int)->float:iflen(bars)<n+1:returnfloat("nan")rets=[bars[i].close/bars[i-1].close-1foriinrange(1,len(bars))]window=rets[-n:]mean=sum(window)/len(window)var=sum((r-mean)**2forrinwindow)/len(window)returnmath.sqrt(var)*math.sqrt(252)defmean_reversion_z(bars:list,n:int)->float:iflen(bars)<n:returnfloat("nan")win=[b.closeforbinbars[-n:]]mean=sum(win)/len(win)sd=math.sqrt(sum((x-mean)**2forxinwin)/len(win))or1e-9return(bars[-1].close-mean)/sdclassFactorEngine:FACTORS={"momentum_20":lambdab:momentum(b,20),"vol_20":lambdab:realized_vol(b,20),"reversion_20":lambdab:mean_reversion_z(b,20)}defcompute(self,source:MarketDataSource,code:str,n:int=50)->dict:bars=source.get_daily_bars(code,n)out={k:round(fn(bars),6)fork,fninself.FACTORS.items()}out["_bars"]=len(bars)returnoutclassCachedDataSource(MarketDataSource):def__init__(self,inner:MarketDataSource):self.inner=inner;self._cache={}defget_daily_bars(self,code:str,n:int)->list:key=(code,n)ifkeyinself._cache:returnself._cache[key]bars=self.inner.get_daily_bars(code,n)self._cache[key]=barsreturnbarsdefmain():eng=FactorEngine()src=MomaDataSource()print("① 真实数据（接口）:",eng.compute(src,"600519",n=50))stub=StubDataSource()r2=eng.compute(stub,"STUB",n=50)print("② 合成数据（零网络）:",r2)assertabs(r2["momentum_20"])<1.0cached=CachedDataSource(MomaDataSource())a=cached.get_daily_bars("600519",50)b=cached.get_daily_bars("600519",50)print("③ 缓存命中幂等:",len(a)==len(b),aisb)if__name__=="__main__":main()

四、跑一遍：同一套因子，两种数据源

真实运行输出（节选）：

解耦演示 ①：真实数据 —— 接口喂同一套因子 600519 因子：{'momentum_20': 0.022747, 'vol_20': 0.136956, 'reversion_20': 0.27838, '_bars': 50} 解耦演示 ②：合成数据 —— 零网络，因子代码一字不改 STUB 因子：{'momentum_20': -0.06709, 'vol_20': 0.122275, 'reversion_20': 0.287604, '_bars': 50} >>> 因子逻辑单测通过（无需任何网络/证书即可验证）。 解耦演示 ③：缓存层 —— 第二次取数不重复打 API（限频友好） 两次取数 bar 数一致：50 == 50 -> True 缓存命中后对象同一性（幂等）：True

看点有两个：

第一，因子函数真的和数据源无关。同样调用eng.compute，喂MomaDataSource拿到 600519 的真实（演示样本）因子，喂StubDataSource拿到合成数据的因子——momentum / realized_vol / reversion_z三个函数的代码一个字没改。这意味着你可以用StubDataSource写单元测试、做边界测试、CI 里零依赖跑通，再切换到MomaDataSource跑真实行情。因子逻辑的正确性，不再依赖「今天接口通不通、证书有没有过期」。

第二，缓存是「装饰器」不是「侵入式」。CachedDataSource只是把任意MarketDataSource包了一层字典缓存，因子引擎毫无感知。第二次取 600519 直接命中缓存、返回同一个对象（a is b），不再打一次接口——对带「次/分钟」频控的套餐尤其友好。同理，你想加重试、熔断、多源对账，都只需在外层再包一层，因子代码永远干净。

五、为什么这套骨架特别适合统一 schema 的行情源

一套稳定且统一 schema 的行情 API 的最大工程优势，是 A 股、ETF、指数同源返回同一套字段（t/o/h/l/c/v/a/pc/sf），归一化成Bar时不用为不同资产品类写分支。配合「证书即取即用、零 SDK 纯 HTTP」的接入方式，你从拿到证书到把数据喂进因子引擎，中间没有任何 SDK 心智负担、没有编译依赖。

这正是解耦能成立的前提：数据源的「接口契约」稳定且统一，你才能在它外面安心地搭缓存、重试、因子库，而不必天天被动适配供应商的私有字段。对比那些字段随时变、接口偶尔失稳的爬虫源，稳定的契约本身就是一种隐性降本。

注意：演示证书返回固定样本，跨股票截面一致属演示 artifact；真实证书下不同标的因子值各异。批量拉数请注意套餐的频控字段，缓存层能显著减少重复请求。

六、小结与下篇预告

本篇把「拉数」和「算因子」用一层接口隔开：纯函数因子 + 因子引擎消费归一化Bar+ 缓存/重试做外层装饰。结果是——因子逻辑可脱离网络单测、换源只改一处、限频靠缓存兜底。数据治理的第一性原理，其实就是「让正确的东西待在正确的层」。

下一篇我们顺着数据治理继续：行情快照的回归测试——用一份「黄金数据集（golden dataset）」每日比对，并把每个数据集版本冻结哈希（hash），确保半年后你能证明「这条回测用的就是这份数据」。

本文仅为技术教程与接口使用示例，文中运行结果来自演示证书在固定样本上的真实执行，量化回测与实盘均有风险，投资需谨慎。A 股惯例：涨幅以红色标注、跌幅以绿色标注。

文档：https://www.momaapi.com/blog/