在加密货币量化交易领域,我们始终关注一个核心问题:价格的驱动力是什么?
为什么有些币能涨那么多?是什么因素驱动了价格上涨?
传统资产比如股票,我们有企业盈利、宏观变量和风险溢价模型,来辅助我们解释资产价格的波动,但加密资产呢?是否存在类似的基本面?
今天要介绍的这篇论文就是专门研究这个问题的。这是一篇发表于 2023 年的研究论文: Do Fundamentals Drive Cryptocurrency Prices?(基本面是否会驱动加密货币的资产定价),它尝试回答一个长期存在的问题:加密货币的价格是否受链上基本面变量驱动,而不仅仅是市场情绪和动量?
在上一篇中,我们对一篇关于市场情绪如何影响资产定价的论文进行了回测,有兴趣的同学可以点这里看完整文章。这次还是老样子,围绕这篇论文构建回测模型,文末会给出完整的Python代码。
论文的原文PDF在这里,建议先看一遍。
论文的核心思想
这篇论文提出,加密货币的价格波动可以由两类区块链层面的基本面因子解释:
- 网络规模因子(gNET)以活跃地址数的增长率衡量,代表网络使用情况和用户基础扩张程度。
- 算力因子(gCP)以 PoW 网络的哈希率增长衡量,反映系统安全性与矿工资源投入。
作者提出这样的假设:是否可以将上面的两点作为加密货币的基本面,然后分析这些基本面对加密货币的价格影响。
然后作者通过 2017–2021 年的链上数据,实证检验了这两个因子是否能解释主要加密货币的横截面收益率。他们发现:
- 活跃地址和算力与价格之间存在显著的长期协整关系;
- 仅使用 gNET 一项,就能解释约 83% 的收益差异;
- 当 gNET 与 gCP 同时加入模型时,拟合度提升至 88%,超过市场、规模、动量等传统因子模型。
换句话说,链上活跃度与资源投入可以视作加密世界的基本面,而非单纯的情绪指标。
策略设计思路
我希望用这篇论文的思路设计一个回测模型,检验这些链上基本面因子在实际交易中的有效性。
在实现上,我构建了一个多因子横截面模型,并以周度为频率进行更新。整体框架如下:
样本范围:2017 年至 2025 年,选取论文中的 18 种主要币种。
数据源:Coin Metrics 的链上数据(活跃地址、算力)与标准 OHLCV 市场数据,实盘交易的话可以使用我们的加密货币实时行情接口。
因子构造:
| gNET | 按周计算活跃地址的 log 一阶差分,1%/99% Winsorize 后等权平均; |
| gCP | 仅对 PoW 币种计算算力增长,同样处理; |
| 因子更新频率 | 每周五 |
| 横截面回归 | 每周五收盘计算各币种收益; 用上期估计的 β 回归横截面收益,估计本期 λ(风险价格); 滚动 156 周窗口计算每个币种对 gNET、gCP 的 β。 |
| 投资组合构建 | 每周一开盘根据 取正向预期收益的币种按比例配置(long-only 模式); 总杠杆 0.9,等风险约束。 |
换句话说,我把论文里的资产定价模型转化为一个动态的量化策略框架,让 λ 和 β 的滚动估计自然形成资产权重信号。
回测结果
| 回测区间 | 2017 – 2025 |
| 夏普比率 | 1.22 |
| 年化收益率 | 65.37% |
| 年化波动率 | 51.88% |
| 最大回撤 | 58.09% |
Python回测代码
以下为完整的回测代码,我们只在Infoway官网发布。如果对您有帮助,请转发给你的朋友。
# -*- coding: utf-8 -*-
"""
Infoway实时行情接口: www.infoway.io
------------------------------------------------------------
实现一个基于 Backtrader框架的策略,逻辑如下:
• 构建周度区块链因子 gNET 和 gCP(以周五到周五为周期),按论文定义计算:gNET 为等权网络增长率,gCP 仅限 PoW 币种,
对每个币的增长率做 1% / 99% Winsorize(去极值处理)。
• 为每个币种估计滚动 156 周的因子暴露(β)。
• 每周通过横截面回归(OLS)估计风险价格 λ,使用上周收益率与上周 β(即 Fama–MacBeth 的时序方式)。
• 每周一开盘再平衡,按预期收益分配权重:E_hat[R_{i,t+1}] = beta_{i,t}^T * lambda_hat_t。
若链上数据集不可用,策略会退化为等权周度再平衡
(同时保持 PWB 框架的必要约束不变)。
参考文献:
• Bhambhwani, S. M., Delikouras, S., & Korniotis, G. M. (2023)。
《Blockchain characteristics and cryptocurrency returns》。
(包含 gNET 与 gCP 定义、周五到周五聚合方式、
Winsorize 处理及滚动 / FMB 节奏)DOI/SSRN: 3342842。
"""
from __future__ import annotations
import math
import re
import numpy as np
import pandas as pd
import pwb_toolbox.backtesting as pwb_bt
import pwb_toolbox.datasets as pwb_ds
# ------------------------------
# 论文中使用的币种集合
# ------------------------------
PAPER_BASELINE_POW = [
"BTC",
"ETH",
"LTC",
"DASH",
"DOGE",
"ETC",
"DCR",
"DGB",
"VTC",
"ZEC",
"XMR",
]
PAPER_BASELINE_NONPOW = ["XRP", "XLM", "LSK", "XEM", "REP", "MAID", "WAVES"]
PAPER_BASELINE_ALL = PAPER_BASELINE_POW + PAPER_BASELINE_NONPOW
# 门罗币的真实活跃地址数据不可观测,根据论文,应在 gNET 计算中排除 XMR
GNET_EXCLUDE = {"XMR"}
def _norm_symbol(sym: str) -> str:
"""
规范化加密货币的符号名称,使其统一格式:
例如 'BTC'、'BTC-USD'、'X:BTCUSD'、'COINBASE:BTC-USD'、'BTCUSDT' -> 'BTC'
仅保留大写英文字母(A–Z),去掉法币或稳定币的后缀。
"""
if sym is None:
return ""
s = sym.upper()
# 去除常见的交易所前缀
if ":" in s:
s = s.split(":")[-1]
# 去除中划线和下划线
s = s.replace("-", "").replace("_", "")
# 去除常见的 USD / USDT / USDC 后缀
s = re.sub(r"(USD|USDT|USDC)$", "", s)
# 仅保留字母部分
s = re.sub("[^A-Z]", "", s)
return s
def _winsorize(series: pd.Series, lower=0.01, upper=0.99) -> pd.Series:
"""通过按分位数裁剪(忽略 NaN)对 Series 进行去极值处理。"""
if series.size == 0:
return series
lo = series.quantile(lower)
hi = series.quantile(upper)
return series.clip(lower=lo, upper=hi)
def _try_load_crypto_dataset(
dataset_names: list[str],
value_candidates: list[str],
date_field_candidates: list[str] = ["date", "datetime"],
) -> pd.DataFrame | None:
"""
尝试按名称加载一个链上数据集,并返回标准化后的 DataFrame,包含:
列:['symbol', 'date', 'value'],其中 'date' 为 UTC 标准化后的 pandas.Timestamp 类型。
该函数会尝试多个数据集名称及字段别名,以提高兼容性。
"""
for ds in dataset_names:
try:
df = pwb_ds.load_dataset(ds)
except Exception:
df = None
if df is None or len(df) == 0:
continue
df_cols = {c.lower(): c for c in df.columns}
# 选取 'symbol' 字段
if "symbol" not in df_cols:
continue
sym_col = df_cols["symbol"]
# 选取日期字段
date_col = None
for c in date_field_candidates:
if c in df_cols:
date_col = df_cols[c]
break
if date_col is None:
continue
# 选取数值字段
val_col = None
for cand in value_candidates:
lc = cand.lower()
if lc in df_cols:
val_col = df_cols[lc]
break
if val_col is None:
# 若当前数据集中没有匹配的候选字段,则跳过
continue
out = df[[sym_col, date_col, val_col]].copy()
out.columns = ["symbol", "date", "value"]
# 若日期为时间戳(整数毫秒),则转换为 UTC 日期;若为 datetime 类型则直接转换
if np.issubdtype(out["date"].dtype, np.number):
out["date"] = pd.to_datetime(out["date"], unit="ms", utc=True)
else:
out["date"] = pd.to_datetime(out["date"], utc=True)
# 规范化符号名称
out["symbol"] = out["symbol"].astype(str).map(_norm_symbol)
# 清洗数值字段
out["value"] = pd.to_numeric(out["value"], errors="coerce").replace(
{np.inf: np.nan, -np.inf: np.nan}
)
out = out.dropna(subset=["symbol", "date"])
return out
return None
def _pivot_daily(df: pd.DataFrame) -> pd.DataFrame:
"""
将整洁格式的数据(symbol, date, value)透视为日频面板数据:
index = 日历日期(UTC,归一化到 00:00),columns = 币种符号,values = 对应数值。
对每个币种进行前向填充(forward fill),以对齐至连续交易日。
"""
if df is None or df.empty:
return pd.DataFrame()
d = df.copy()
# 将日期标准化为 UTC 格式
d["date"] = d["date"].dt.tz_convert("UTC").dt.normalize()
panel = d.pivot_table(
index="date", columns="symbol", values="value", aggfunc="last"
).sort_index()
# 新索引为连续的日历日期,并前向填充
full_idx = pd.date_range(
start=panel.index.min(), end=panel.index.max(), freq="D", tz="UTC"
)
panel = panel.reindex(full_idx).ffill()
return panel
def _weekly_diff_from_daily(panel: pd.DataFrame) -> pd.Series:
"""
从日频宽表数据中计算每个币种的周度(周五到周五)对数一阶差分,然后对每周的所有币种取等权平均,得到周度因子。
步骤(与论文一致):
1)提取每周五(W-FRI)的最后一个观测值;
2)取自然对数后计算一阶差分 Δ;
3)对每个币种的 Δ 进行 1% / 99% Winsorize 去极值;
4)每周对所有可用币种的增长率取等权平均。
返回:pd.Series(索引为周五日期,值为等权平均增长率)。
"""
if panel is None or panel.empty:
return pd.Series(dtype="float64")
# 周度采样(取每周五的观测)
weekly = panel.resample("W-FRI").last()
# 取对数变换
weekly = np.log(weekly.replace({0.0: np.nan}))
# 对每列计算一阶差分
delta = weekly.diff()
# 对每个币种(按列)去极值
for col in delta.columns:
delta[col] = _winsorize(delta[col], 0.01, 0.99)
# 对每周所有币种取等权平均(按行求均值)
eq = delta.mean(axis=1, skipna=True)
return eq.dropna()
class PaperStrategy(pwb_bt.BaseStrategy):
"""
基于 Bhambhwani, Delikouras & Korniotis (2023) 的链上因子横截面模型,
实现的 Backtrader 策略类。
参数说明:
• leverage:投资组合总杠杆上限(默认为多头模式)。
• beta_window_weeks:计算 β 的滚动窗口长度(默认 156 周)。
• min_weeks:计算 β 所需的最小周数(默认 52 周)。
• lambda_mode:风险价格 λ 的估计方式,可选:
- 'rolling_fmb'(默认):每周五基于上周收益与上周 β 估计 λ_t;
- 'paper_constants':使用论文给出的常数 λ(gNET=0.098, gCP=0.033)。
• long_only:若为 True,则将负的预期收益置零后再归一化权重。
再平衡频率:每周一开盘,根据截至上周五的信息进行调仓。
"""
params = (
("leverage", 0.90),
("total_days", 0),
("indicator_cls", None),
("indicator_kwargs", {}),
# 策略专用参数(安全默认值)
("beta_window_weeks", 156),
("min_weeks", 52),
("lambda_mode", "rolling_fmb"),
("lambda_net_paper", 0.098), # 表 5:单因子 gNET(全样本)
("lambda_cp_paper", 0.033), # 表 5:单因子 gCP(全样本)
("long_only", True),
)
# ------------------------------
# 回测框架生命周期
# ------------------------------
def __init__(self):
super().__init__()
# 将 Backtrader 的数据源映射为标准化后的币种符号
self._data_by_sym = {}
for d in self.datas:
# 优先使用 _name(Backtrader 的名称),否则使用 dataname 的字符串
nm = getattr(d, "_name", None) or str(
getattr(d, "params", getattr(d, "p", None)).dataname
)
sym = _norm_symbol(nm)
self._data_by_sym[sym] = d
self._symbols_universe = sorted(self._data_by_sym.keys())
# --- 加载链上数据集(活跃地址与算力) ---
# 活跃地址(代表网络规模)
net_raw = _try_load_crypto_dataset(
dataset_names=[
"Crypto-Daily-ActiveAddresses",
"CoinMetrics-ActiveAddresses",
"All-Active-Addresses",
],
value_candidates=[
"unique_active_addresses",
"active_addresses",
"active_address",
"active",
],
)
self._net_daily = (
_pivot_daily(net_raw) if net_raw is not None else pd.DataFrame()
)
# 算力(代表计算能力)
cp_raw = _try_load_crypto_dataset(
dataset_names=[
"Crypto-Daily-Hashrate",
"CoinMetrics-Hashrate",
"All-Hashrate",
],
value_candidates=[
"hashrate",
"hash_rate",
"hash_rate_7d",
"hash_rate_7d_avg",
],
)
self._cp_daily = _pivot_daily(cp_raw) if cp_raw is not None else pd.DataFrame()
# 确定可用于因子构建的币种
net_syms = (
set(map(_norm_symbol, self._net_daily.columns))
if not self._net_daily.empty
else set()
)
cp_syms = (
set(map(_norm_symbol, self._cp_daily.columns))
if not self._cp_daily.empty
else set()
)
# 根据论文设定:gNET 使用基准样本(排除 XMR);gCP 仅使用 PoW 币种
gnet_syms = [
s for s in PAPER_BASELINE_ALL if (s in net_syms and s not in GNET_EXCLUDE)
]
gcp_syms = [s for s in PAPER_BASELINE_POW if (s in cp_syms)]
# 按币种子集提取面板数据并计算周度因子
if not self._net_daily.empty and gnet_syms:
gnet_panel = self._net_daily[
[c for c in self._net_daily.columns if _norm_symbol(c) in gnet_syms]
]
self._gnet_weekly = _weekly_diff_from_daily(gnet_panel)
else:
self._gnet_weekly = pd.Series(dtype="float64")
if not self._cp_daily.empty and gcp_syms:
cp_panel = self._cp_daily[
[c for c in self._cp_daily.columns if _norm_symbol(c) in gcp_syms]
]
self._gcp_weekly = _weekly_diff_from_daily(cp_panel)
else:
self._gcp_weekly = pd.Series(dtype="float64")
self._factors_ok = (len(self._gnet_weekly) > 0) and (len(self._gcp_weekly) > 0)
# 历史数据记录(按周)
self._week_dates: list[pd.Timestamp] = []
self._ret_hist: dict[str, list[float]] = {s: [] for s in self._symbols_universe}
self._gnet_hist: list[float] = []
self._gcp_hist: list[float] = []
# 滚动 β(用于下一期 λ 的估计)
self._betas_prev: dict[str, np.ndarray] = {
s: np.array([0.0, 0.0]) for s in self._symbols_universe
}
# 以最新窗口计算 β(用于下一周)
self._betas_next: dict[str, np.ndarray] = {
s: np.array([0.0, 0.0]) for s in self._symbols_universe
}
# 初始化 λ(风险价格)
if self.p.lambda_mode == "paper_constants":
self._lambda = np.array(
[self.p.lambda_net_paper, self.p.lambda_cp_paper], dtype=float
)
else:
self._lambda = np.array([0.0, 0.0], dtype=float)
# 记录每个币种上周五收盘价以计算周度收益
self._last_fri_close: dict[str, float | None] = {
s: None for s in self._symbols_universe
}
# 内部标志位,用于触发下周一的再平衡操作
self._rebalance_ready = False
def prenext(self):
# Backtrader预热;此处无需特殊逻辑。
pass
def next(self):
super().next() # 勿删除
# 当前 K 线的时间(naive 时间即可;Backtrader 使用交易所时区)
dt = pd.Timestamp(self.datas[0].datetime.datetime(0)).tz_localize(
"UTC", nonexistent="shift_forward"
)
# ------------------------------------------------------------------
# 1) 1)每个周五收盘:计算上周的收益与因子,
# 基于前一期的 β 估计 λ(风险价格),然后更新新的 β。
# ------------------------------------------------------------------
if dt.weekday() == 4:
# 1a) 计算每个币种的周五收盘价及周度收益(周五到周五)
week_ret = {}
for s in self._symbols_universe:
d = self._data_by_sym[s]
close = float(d.close[0])
last = self._last_fri_close[s]
if last is not None and last > 0:
r = (close / last) - 1.0
week_ret[s] = r
# 更新上周五的收盘价记录
self._last_fri_close[s] = close
# 若为首个周五无历史数据或因子缺失,则跳过
if (not week_ret) or (not self._factors_ok):
# 仍设置再平衡标志,以便周一可执行等权再平衡
self._rebalance_ready = True
return
# 1b) 提取本周五的 gNET / gCP 值(即本周 t 的因子实现值)
# 同步至周五 00:00(UTC)以进行索引匹配
fri = dt.normalize()
if fri not in self._gnet_weekly.index or fri not in self._gcp_weekly.index:
# 若因子数据中无当前周五(极少发生),则取最近的前一周五数据
gnet_t = (
self._gnet_weekly[self._gnet_weekly.index <= fri].iloc[-1]
if len(self._gnet_weekly)
else np.nan
)
gcp_t = (
self._gcp_weekly[self._gcp_weekly.index <= fri].iloc[-1]
if len(self._gcp_weekly)
else np.nan
)
else:
gnet_t = float(self._gnet_weekly.loc[fri])
gcp_t = float(self._gcp_weekly.loc[fri])
if (not np.isfinite(gnet_t)) or (not np.isfinite(gcp_t)):
# 若本周因子数据不可用,则跳过更新,但允许执行等权再平衡
self._rebalance_ready = True
return
# 1c) 将数据追加至历史记录
self._week_dates.append(fri)
for s in self._symbols_universe:
self._ret_hist[s].append(week_ret.get(s, np.nan))
self._gnet_hist.append(gnet_t)
self._gcp_hist.append(gcp_t)
# 1d) 若已有前期 β,则估计 λ(横截面回归 OLS)
# r_t = alpha + B_{t-1} * lambda_t + noise
if self.p.lambda_mode == "rolling_fmb":
# 收集具有有限收益率 r_t 和有效 β 的资产
Y = []
X = []
for s in self._symbols_universe:
r_t = self._ret_hist[s][-1] if self._ret_hist[s] else np.nan
b = self._betas_prev.get(s, None)
if b is None or not np.all(np.isfinite(b)):
continue
if r_t is None or not np.isfinite(r_t):
continue
Y.append(r_t)
X.append(b.tolist())
if (
len(Y) >= 3
): # 至少需要 3 个资产才能拟合含两个系数和截距的模型
Y = np.asarray(Y)
X = np.asarray(X)
# 加入截距项
Z = np.column_stack([np.ones(len(Y)), X])
try:
coef = np.linalg.lstsq(Z, Y, rcond=None)[0]
# 回归系数含义:coef = [α, λ_net, λ_cp]
self._lambda = coef[1:].astype(float)
except Exception:
# 若回归失败,则保留上一期的 λ 不变
pass
# 1e) 使用最近 W 周(含当前周 t)的数据重新计算 β
L = len(self._gnet_hist)
window = self.p.beta_window_weeks
start = max(0, L - window)
# 准备因子矩阵 F(不包含截距项)
F_net = np.asarray(self._gnet_hist[start:L], dtype=float)
F_cp = np.asarray(self._gcp_hist[start:L], dtype=float)
F = np.column_stack([F_net, F_cp]) # shape: [T, 2]
# 对每个资产,以相同时间窗口内的收益回归于因子矩阵 F
for s in self._symbols_universe:
y = np.asarray(self._ret_hist[s][start:L], dtype=float)
mask = np.isfinite(y) & np.isfinite(F_net) & np.isfinite(F_cp)
if mask.sum() >= max(
self.p.min_weeks, 10
): # 确保样本数量足够
# 加入截距项
Xmat = np.column_stack([np.ones(mask.sum()), F[mask]])
try:
beta = np.linalg.lstsq(Xmat, y[mask], rcond=None)[0]
# 回归系数:beta = [α_i, β_net_i, β_cp_i]
self._betas_next[s] = beta[1:].astype(float)
except Exception:
# 若计算失败,则保留上一期的 β
pass
# 否则保持原值不变
# 1f) 将 β 向前滚动至下一周,并设置周一再平衡标志
self._betas_prev = {
s: self._betas_next.get(s, np.array([0.0, 0.0]))
for s in self._symbols_universe
}
self._rebalance_ready = True
# ------------------------------------------------------------------
# 2) 每周一开盘:根据上周五的 λ 与 β 进行再平衡
# ------------------------------------------------------------------
if dt.weekday() == 0 and self._rebalance_ready:
# 计算每个资产的预期收益得分:score_i = β_i^T * λ
scores = {}
for s in self._symbols_universe:
b = self._betas_prev.get(s, np.array([0.0, 0.0]))
if b is None or not np.all(np.isfinite(b)):
continue
sc = float(np.dot(b, self._lambda))
scores[s] = sc
# 若因子或 β 缺失,则使用等权配置作为默认方案
if (not scores) or (
all((not np.isfinite(v)) or v == 0.0 for v in scores.values())
):
weights = {
s: self.p.leverage / max(1, len(self._symbols_universe))
for s in self._symbols_universe
}
else:
# 执行多头或多空权重归一化
vals = np.array(
[scores[s] for s in self._symbols_universe], dtype=float
)
if self.p.long_only:
vals = np.where(np.isfinite(vals) & (vals > 0), vals, 0.0)
if vals.sum() <= 0:
vals = np.ones_like(vals)
raw_w = vals / vals.sum()
raw_w *= self.p.leverage
else:
# 对于多空组合:去均值以保持美元中性,再按 L1 范数缩放至目标杠杆
vals = np.where(np.isfinite(vals), vals, 0.0)
vals = vals - vals.mean()
l1 = np.abs(vals).sum()
if l1 <= 0:
raw_w = np.zeros_like(vals)
else:
raw_w = (self.p.leverage * vals) / l1
weights = {s: float(w) for s, w in zip(self._symbols_universe, raw_w)}
# 发送交易指令
for s, w in weights.items():
d = self._data_by_sym[s]
try:
# Backtrader 框架提供 order_target_percent 方法用于下单
self.order_target_percent(data=d, target=w)
except Exception:
pass
# 清除再平衡标志
self._rebalance_ready = False
# ------------------------------
# 执行入口
# ------------------------------
def run_strategy(
symbols: list[str] | None = None,
start_date: str = "2017-01-06",
cash: float = 100_000.0,
leverage: float = 0.90,
lambda_mode: str = "rolling_fmb", # or 'paper_constants'
long_only: bool = True,
):
"""
参数说明
----------
symbols : list[str]
币种列表(每日 OHLCV 数据)。若为 None,则使用论文中的
18 个基准币种(符号与 Coin Metrics / 常见交易所一致)。
start_date : str
回测开始日期(默认为 '2017-01-06',与论文一致)。
cash : float
初始资金。
leverage : float
投资组合的最大杠杆(适用于多头分配)。
lambda_mode : str
风险价格 λ 的模式:
'rolling_fmb':按周估计 λ;
'paper_constants':使用论文给定的固定 λ(λ_gNET=0.098, λ_gCP=0.033)。
long_only : bool
是否仅做多(默认 True),或允许多空交易。
返回值
-------
"""
if symbols is None:
symbols = PAPER_BASELINE_ALL.copy()
return pwb_bt.run_strategy(
strategy_cls=PaperStrategy,
strategy_kwargs={
"leverage": leverage,
"lambda_mode": lambda_mode,
"long_only": long_only,
},
indicator_cls=None,
indicator_kwargs={},
symbols=symbols,
start_date=start_date,
cash=cash,
)
if __name__ == "__main__":
strategy = run_strategy()