在开发量化模型时，我们经常试图从非传统因子中提取市场信号，而投资者情绪无疑是其中最具争议也最有潜力的方向之一。今天要介绍的研究正是 Anchada Charoenrook的经典论文《Does Sentiment Matter?》，这篇论文系统地研究了消费者情绪变化与市场超额收益之间的关系。

这篇论文首次发表于2006年，为了验证它今天还有没有用，我们根据论文里的核心观点写了一段Python，并放到2023-2025这段时间里跑一跑，看看最终的效果如何。我强烈建议你先看看论文的原文，PDF文档放在这里。

最后我们会给出完整的回测代码，希望对你有帮助。

更多精彩内容：

我们还是先回过头来讲讲这篇论文说了些啥。

论文核心观点

论文的核心问题是：投资者情绪是否会影响市场回报？

作者使用了密歇根大学的 Consumer Sentiment Index（CSI） 作为情绪代理变量，并定义了其年度变化率（即 CCSI = ΔCSI / CSIₜ₋₁）。研究发现：

  • 当消费者情绪上升（投资者过于乐观）时，未来一个月及未来一年内的超额市场回报显著下降；
  • 当情绪下降（投资者悲观）时，未来市场回报反而更高；
  • 这种预测能力在统计与经济意义上都显著，并且独立于其他宏观控制变量（如股息率、利差、消费–财富比等）；
  • 情绪变化比传统的预测因子（如股息收益率）更能解释未来的市场走势。

换句话说，投资者的情绪变化本身就是一种系统性信号。

回测思路

我基于该论文搭建了一个情绪择时模型，并用 Python 实现了完整的回测逻辑。核心思路如下：

1. 信号构建

从 UMich 消费者信心指数中计算出同比变化率 CCSI，并滞后两个月（对应论文中的信息发布延迟处理）。

\begin{equation} CCSI_t = \frac{CSI_t – CSI_{t-12}}{CSI_{t-12}}, \quad Signal = CCSI_{t-2} \end{equation}

2. 回归参数引用

使用论文 Table 4 的月度回归系数：

\begin{equation} \hat{r}_t = a + b \times CCSI_{t-1}, \quad a = 0.009, \ b = -0.049 \end{equation}

当预测的超额收益\( \hat{r}_t > 0 \)时，持有市场风险资产；否则持币观望。

3. 交易规则

调仓频率：每月第一个交易日

仓位分配：等权持仓

杠杆设定：0.9

资产标的：S&P500 ETF（SPY）作为市场代表。

这种逻辑非常直接，没有多因子，也没有复杂的优化。它几乎是一个纯学术信号的落地测试，目的是验证论文结果在现代市场数据中的延续性。

回测结果

回测区间2023 – 2025
夏普比率1.6
年化收益率20.42%
年化波动率12.06%
最大回撤11.36%

Python代码

Python
# sentiment_timing.py
# Infoway实时行情接口:www.infoway.io
# 原文链接：https://blog.infoway.io/python-sentiment-backtest/

import math
import datetime as dt
from collections import defaultdict

import numpy as np
import pandas as pd

import pwb_toolbox.backtesting as pwb_bt
import pwb_toolbox.datasets as pwb_ds


# --- 辅助函数：日期时间解析 ----------------------------------------


def _parse_datetimes_flexible(col: pd.Series) -> pd.Series:
    """
    Return a UTC tz-aware datetime Series from a column that may be:
    - epoch in s / ms / us / ns
    - ISO 8601 strings
    - mixed types
    """
    s = pd.Series(col).copy()

    # 优先尝试将数据解析为时间戳
    num = pd.to_numeric(s, errors="coerce")
    if num.notna().mean() > 0.5:  # majority numeric? try epoch units
        # 取一小部分样本，通过合理性判断推断时间戳单位
        sample = num.dropna().astype("int64")
        sample = sample.iloc[: min(1000, len(sample))]

        def plausible(ts: pd.Series) -> bool:
            yrs = ts.dt.year
            # 若有 90% 以上的时间点落在现代区间（1950～2100 年），则认为单位有效
            return (yrs.between(1950, 2100)).mean() >= 0.9

        for unit in ("s", "ms", "us", "ns"):
            try:
                ts_try = pd.to_datetime(sample, unit=unit, utc=True)
                if plausible(ts_try):
                    # 使用推断出的时间单位转换整列数据
                    return pd.to_datetime(num.astype("Int64"), unit=unit, utc=True)
            except Exception:
                pass

        # 若失败则回退到毫秒级解析
        return pd.to_datetime(num.astype("Int64"), unit="ms", utc=True, errors="coerce")

    # 否则按字符串格式（如 ISO 格式）进行解析
    return pd.to_datetime(s, utc=True, errors="coerce")


def _first_existing(df: pd.DataFrame, candidates) -> str:
    for c in candidates:
        if c in df.columns:
            return c
    raise KeyError(f"No datetime-like column found; tried: {candidates}")


def _try_load_umich_csi():
    """
    尝试加载几种可能包含UMich CSI的数据集名称
    若找到则返回一个包含 ['date_ms', 'CSI'] 的 pandas DataFrame，否则返回 None
    """
    candidate_names = [
        "Macro-UMich-ConsumerSentiment",
        "Macro-UMich-CSI",
        "Macro-UMich-ICS",
        "Macro-Consumer-Sentiment-UMich",
    ]
    for name in candidate_names:
        try:
            df = pwb_ds.load_dataset(name)
            if df is None or len(df) == 0:
                continue
            df = df.copy()
            # 通过启发式方法定位 CSI 指标列
            col_candidates = [
                c
                for c in df.columns
                if c.lower() in {"csi", "ics", "index", "value", "level"}
            ]
            if not col_candidates:
                continue
            vcol = col_candidates[0]
            # 统一字段格式
            if "date" in df.columns:
                date_col = "date"
            elif "datetime" in df.columns:
                date_col = "datetime"
            else:
                # 若没有日期类型的列则放弃本次尝试
                continue
            out = pd.DataFrame(
                {
                    "date_ms": df[date_col].astype("int64"),
                    "CSI": pd.to_numeric(df[vcol], errors="coerce"),
                }
            ).dropna()
            return out
        except Exception:
            # 续尝试下一个候选数据集；库环境中无需记录日志
            pass
    return None


# --- 用作 _build_ccsi_series 的替代实现 -------------------------------
def _build_ccsi_series(source="auto", epsilon=0.1):
    """
    构建消费者情绪的同比（月度）百分比变化（CCSI），并加入论文中的两个月滞后
    返回：
        ccsi_lag2 : pd.Series (索引为月末 UTC 时间，值为浮点型)
        used_source : str  ("umich" | "news")
    """
    used_source = None

    # 1) 如果可以获取 UMich CSI 数据，优先使用
    umich = None
    if source in ("auto", "umich"):
        umich = _try_load_umich_csi()

    if umich is not None:
        used_source = "umich"
        s = (
            umich.assign(ts=pd.to_datetime(umich["date_ms"], unit="ms", utc=True))
            .set_index("ts")
            .sort_index()["CSI"]
            .astype(float)
        )
        s_m = s.resample("M").last().dropna()
        ccsi = (s_m - s_m.shift(12)) / s_m.shift(12)

    else:
        # 2) 否则回退到基于 All-Daily-News 数据构建的宏观情绪指标
        used_source = "news"
        news_df = pwb_ds.load_dataset("All-Daily-News")
        df = news_df.copy()

        # 寻找时间戳字段
        dt_col = _first_existing(
            df,
            ["datetime", "published_utc", "published_at", "time", "timestamp", "date"],
        )
        df["ts"] = _parse_datetimes_flexible(df[dt_col])

        # 情绪值可能是字符串，需要转换为浮点型
        if "sentiment" not in df.columns:
            raise KeyError("All-Daily-News has no 'sentiment' column.")
        df["sentiment"] = pd.to_numeric(df["sentiment"], errors="coerce")

        df = df.dropna(subset=["ts", "sentiment"])
        if df.empty:
            raise ValueError("All-Daily-News provided no usable rows after parsing.")

        # 按月计算平均情绪
        m = df.set_index("ts")["sentiment"].resample("M").mean().dropna()

        # 计算同比百分比变化，并加入一个微小的稳定项
        ccsi = (m - m.shift(12)) / (m.shift(12).abs() + float(epsilon))

    # 按论文设定，使用 t-2 的 CCSI 作为当月信号
    ccsi_lag2 = ccsi.shift(2).dropna()
    return ccsi_lag2, used_source


def _month_key(ts):
    """从pandas或datetime对象中返回 (year, month)，（支持 UTC 或普通时间）"""
    if isinstance(ts, (pd.Timestamp, np.datetime64)):
        ts = pd.Timestamp(ts).to_pydatetime()
    return ts.year, ts.month


class PaperStrategy(pwb_bt.BaseStrategy):
    """
    基于论文《Does Sentiment Matter?》的情绪择时策略实现

    核心逻辑（源自论文）：
      - 信号（Signal）：CCSI_t 表示密歇根消费者信心指数（UMich CSI）的同比变化。
      - 时序（Timing）：将当月收益与两个月前的 CCSI 对齐（滞后两个月，以符合数据发布时间）。
      - 频率（Cadence）：按月执行（每月第一个交易日进行调仓）。
      - 决策（Decision）：使用论文中单变量（月度、加权平均）回归系数预测超额收益：
              r_hat_t = a + b * CCSI_{t-1}，其中 a=0.009，b=-0.049（见表4 Panel A）。
        若预测超额收益 r_hat_t > 0 → 增加风险敞口（全仓做多，等权分配到所有标的）； 否则 → 保持现金仓位。

    说明:
      - 论文研究的是市场整体的超额收益，这里将同样的情绪信号应用于所有输入的标的（通常是宽基指数 ETF，如 'SPY'）。
      - 年度版本（从7月到次年6月）的回归系数见表5 Panel A（a=0.079，b=-0.452），
        代码中已保留该选项，但默认关闭，以保持与月度测试结果一致。
    """

    params = (
        ("leverage", 0.9),
        ("total_days", 0),  # 保留自模板代码
        ("use_one_year_horizon", False),
        ("threshold", 0.0),  # 仅当预测值大于阈值时才开多仓
        ("ccsi_source", "auto"),
        # 论文中的回归系数（按月度加权平均）—— 表4 Panel A：
        ("intercept_a", 0.009),
        ("slope_b", -0.049),
        # 年度回归系数（加权平均）—— 表5 Panel A（仅在 use_one_year_horizon=True 时启用）
        ("annual_intercept_a", 0.079),
        ("annual_slope_b", -0.452),
        # PWB 框架所需参数（请勿删除）
        ("indicator_cls", None),
        ("indicator_kwargs", {}),
    )

    def __init__(self):
        super().__init__()

        # 构建一个字典，将每个 (year, month) 映射到对应的 CCSI_{t-1}，包含论文要求的两个月滞后
        ccsi_series, used_source = _build_ccsi_series(self.p.ccsi_source)
        self._ccsi_source = used_source

        # 为 next() 方法提供快速查询：{(year, month): float}
        self._ccsi_by_month = {_month_key(ix): val for ix, val in ccsi_series.items()}

        # 记录调仓状态
        self._last_rebalanced_key = None

        # 便捷属性
        self._nd = len(self.datas)
        self._equal_weight = self.p.leverage / max(1, self._nd)

        # 保存当前时间周期（如月度或年度）对应的回归系数
        if self.p.use_one_year_horizon:
            self._a = self.p.annual_intercept_a
            self._b = self.p.annual_slope_b
        else:
            self._a = self.p.intercept_a
            self._b = self.p.slope_b

    # --- 工具方法 -----------------------------------------------------

    def _is_first_trading_day_of_month(self):
        d0 = self.datas[0]
        if len(d0) == 0:
            return False
        today = d0.datetime.date(0)
        if len(d0) <= 1:
            # 回测中的第一根K线视为“当月首个交易日”
            return True
        yesterday = d0.datetime.date(-1)
        return (today.month != yesterday.month) or (today.year != yesterday.year)

    def _current_month_key(self):
        d0 = self.datas[0]
        today = d0.datetime.datetime(0)  # 返回datetime
        return today.year, today.month

    def _forecast_excess_return(self, ccsi_value):
        # 按论文中的单变量回归公式计算线性预测值
        if ccsi_value is None or (
            isinstance(ccsi_value, float) and not math.isfinite(ccsi_value)
        ):
            return None
        return float(self._a + self._b * ccsi_value)

    # --- 交易逻辑 -------------------------------------------------------

    def next(self):
        super().next()

        # 仅在每月第一个交易日进行决策
        if not self._is_first_trading_day_of_month():
            return

        mkey = self._current_month_key()
        if self._last_rebalanced_key == mkey:
            return  

        # 获取与当前收益月份对应的滞后 CCSI（论文中为 t 使用 t-2）
        ccsi = self._ccsi_by_month.get(mkey, None)
        forecast = self._forecast_excess_return(ccsi)

        # 若尚无可用的 CCSI 数据，则当月不操作
        if forecast is None:
            return

        take_risk = forecast > self.p.threshold

        # 目标仓位：若判断应承担风险则等权持仓，否则全仓持币
        target_pct = self._equal_weight if take_risk else 0.0

        for data in self.datas:
            self.order_target_percent(data=data, target=target_pct)

        self._last_rebalanced_key = mkey


# --- 策略运行模板 ------------------------------------------------------


def run_strategy():
    """
    示例运行函数。
    - 传入指数（如 'SPY'）；情绪择时信号会统一应用到所有标的上。
    - 起始日期需保证有足够的数据预热，用于计算同比变化和两个月滞后。
    """
    symbols = [
        "SPY",  
        # 也可以添加其他ETF，信号会对所有标的等效生效
    ]
    return pwb_bt.run_strategy(
        strategy_cls=PaperStrategy,
        strategy_kwargs={"leverage": 0.9},  # <-- 按模板保留此结构
        indicator_cls=None,  # PWB 框架保留项
        indicator_kwargs={},  # PWB 框架保留项
        symbols=symbols,  # 使用日K数据
        start_date="2023-06-01",  # 与论文所用的长期历史数据保持一致
        cash=100_000.0,
    )


if __name__ == "__main__":
    strategy = run_strategy()