计算基金日频因子

安装 DolphinDB server，配置为单节点模式，其默认端口号为8848。本次测试所涉及到的硬件环境和软件环境如下：

• 硬件环境

  硬件名称	配置信息
  操作系统	64位 CentOS Linux 7 (Core)
  内存	256 GB
  CPU 类型	Intel(R) Xeon(R) Silver 4216 CPU @ 2.10GHz

• 软件环境

  软件名称	版本信息	教程&下载链接
  DolphinDB	2.00.7	DolphinDB单节点部署教程；DolphinDB下载链接
  DolphinDB GUI	1.30.19.2	DolphinDB GUI下载链接
  Python（包含numpy、pandas、scipy等库以及 DolphinDB 的 Python API，需要预先安装）	3.7.9	Python官网

本教程选取了10个基金日频因子，涵盖基金评价体系的各个方面，作为性能测试的指标。其相关的背景含义、计算公式及在 DolphinDB 脚本和 Python 中的定义如下所示：

注意：下文提到的 DailyValue 指日净值，DailyReturn 指日收益率，BeginningValue 指基金日净值序列中的第一个净值数据，EndingValue 指基金日净值序列中的最后一个净值数据。

年化收益率

• 因子含义：将当前收益率转换为一年内（按252个交易日计算）的收益率；

• 公式：

  
  
  

代码实现：

• DolphinDB：

      defg getAnnualReturn(value){
            return pow(1 + ((last(value) - first(value))\first(value)), 252\730) - 1
      }

• Python：

      def getAnnualReturn(value):
          return pow(1 + ((value[-1] - value[0])/value[0]), 252/730)-1

年化波动率

• 因子含义：衡量基金的波动风险

• 公式：

  
  
  

代码实现：

• DolphinDB：

        defg getAnnualVolatility(value){
        	return std(deltas(value)\prev(value)) * sqrt(252)
        }

• Python：

        def getAnnualVolatility(value):
              diff_value = np.diff(value)
              rolling_value = np.roll(value, 1)
            rolling_value = np.delete(rolling_value, [0])
            return np.std(np.true_divide(diff_value, rolling_value), ddof=1) * np.sqrt(252)

收益率偏度

• 因子含义：衡量收益偏斜程度

• 公式：

  
  
  

代码实现：

• DolphinDB：

      defg getAnnualSkew(value){
      	return skew(deltas(value)\prev(value))
      }

• Python：

      def getAnnualSkew(value):
          diff_value = np.diff(value)
          rolling_value = np.roll(value, 1)
          rolling_value = np.delete(rolling_value, [0])
          return st.skew(np.true_divide(diff_value, rolling_value))

收益率峰度

• 因子含义：基金日收益率的分布相对于正态分布的峰值程度和尾部厚度

• 公式：

  
  
  

代码实现：

• DolphinDB：

      defg getAnnualKur(value){
      	return kurtosis(deltas(value)\prev(value)) 
      }

• Python：

      def getAnnualKur(value):
          diff_value = np.diff(value)
          rolling_value = np.roll(value, 1)
          rolling_value = np.delete(rolling_value, [0])
          return st.kurtosis(np.true_divide(diff_value, rolling_value), fisher=False)

夏普比率

• 因子含义：反映单位风险基金净值增长率超过无风险收益率的程度

• 公式：

  
  
  

代码实现：

• DolphinDB：

      defg getSharp(value){
      	return (getAnnualReturn(value) - 0.03)\getAnnualVolatility(value) as sharpeRat
      }

• Python：

      def getSharp(value):
          return (getAnnualReturn(value) - 0.03)/getAnnualVolatility(value) if getAnnualVolatility(value) != 0 else 0

最大回撤率

• 因子含义：指在选定周期内任一历史时点往后推，产品净值走到最低点时的收益率回撤幅度的最大值

• 公式：在某个时间周期的净值序列中，记 i 为某一天，DailyValuei 为第 i 天的净值，j 为 i 之前的某一天，DailyValuej 为第 j 天的净值，找到一组 i, j 使得其收益率回撤幅度最大记为最大回撤率 MaxDrawdown，

  
  
  

代码实现：

      def getMaxDrawdown(value){
      	i = imax((cummax(value) - value) \ cummax(value))
      	if (i==0){
      		return 0
      	}
      	j = imax(value[:i])
      	return (value[j] - value[i]) \ (value[j])
      }

• Python：

      def getMaxDrawdown(value):
          i = np.argmax((np.maximum.accumulate(value) - value) / np.maximum.accumulate(value))
          if i == 0:
              return 0
          j = np.argmax(value[:i])
          return (value[j] - value[i]) / value[j]

收益回撤比

• 因子含义：是收益和风险的比值，可以用来衡量一只基金产品投资策略的好坏。

• 公式：

  
  
  

代码实现：

• DolphinDB：

      def getDrawdownRatio(value){
      	return getAnnualReturn(value) \ getMaxDrawdown(value)
      }

• Python：

      def getDrawdownRatio(value):
          return getAnnualReturn(value) / getMaxDrawdown(value) if getMaxDrawdown(value) != 0 else 0

β系数

• 因子含义：一种风险指数，可以衡量股票基金相对于整个股市的波动情况

• 公式：

  
  
  

其中分子为日净值收益率和日基准收益率的协方差，分母为日基准收益率的方差

代码实现：

• DolphinDB：

      def getBeta(value, price){
      	return covar(deltas(value)\prev(value), deltas(price)\prev(price)) \ std(deltas(price)\prev(price))
      }

• Python：

      def getBeta(value, price):
          diff_price = np.diff(price)
          rolling_price = np.roll(price, 1)
          rolling_price = np.delete(rolling_price, [0])
          diff_value = np.diff(value)
          rolling_value = np.roll(value, 1)
          rolling_value = np.delete(rolling_value, [0])
          return np.cov(np.true_divide(diff_value, rolling_value), np.true_divide(diff_price, rolling_price))[0][1] / np.std(np.true_divide(diff_price, rolling_price), ddof=1)

α系数

• 因子含义：一种风险指数，用于衡量与市场波动无关的超额收益

• 公式：

  
  
  

代码实现：

• DolphinDB：

      def getAlpha(value, price){
      	return getAnnualReturn(value) - 0.03 - getBeta(value, price) * (getAnnualReturn(price) - 0.03)
      }

• Python：

      def getAlpha(value, price):
          return getAnnualReturn(value) - 0.03 - getBeta(value, price) * (getAnnualReturn(price) - 0.03)

赫斯特指数

• 因子含义：体现时间序列的自相关性，尤其是序列中隐藏的长期趋势的指标。此处用来揭示某只基金序列中的隐藏长期趋势。

• 计算方法：

  a. 将收益率序列按照不同的粒度（记为 k）切分成不同的片段，此处从粒度2到粒度365开始切分序列； b. 计算每个片段的均值，此处要计算2+3+4+...+365种；

• 公式：

  
  
  

c. 计算每个片段的离差序列，即用每个片段的均值分别减该片段内的每个元素；

• 公式：

  
  
  

d. 计算每个片段离差序列中的最大值减去最小值，记为 R；

• 公式：

  
  
  

e. 计算每个片段的标准差，记为 S；

• 公式：

  
  
  

f. 计算每个片段的 R/S 值，并计算按照不同粒度划分的所有片段中各片段元素的均值；

• 公式：

  
  
  

g. 以划分的标准粒度为自变量，以对应的各片段元素的均值为因变量，计算回归方程，截距即为赫斯特指数；

• 公式：AVS=a*k+b，经过计算后系数 a 为赫斯特指数

代码实现：

• DolphinDB：

        def calAllRs2(mret, symList, k){
                rowCount = mret.rows()/k * k
                demeanCum = rolling(cumsum, mret[0:rowCount,] - each(stretch{, rowCount}, rolling(avg, mret, k, k)), k, k)
                a = rolling(max, demeanCum, k, k) - rolling(min, demeanCum, k, k)
                RS = nullFill!(a/rolling(stdp, mret, k, k), 1.0).mean().log()
                return table(symList as fundNum, take(log(k), symList.size()) as knum, RS as factor1)
        }

• Python：

        def calHurst(value_list, min_k):
            n = len(value_list)
            max_k = int(np.floor(n / 2))
            r_s_dict = []
            for k in range(min_k, max_k +1):
                subset_list = [value_list[i: i+k] for i in range(0, n, k)]
                if np.mod(n, k) > 0:
                    subset_list.pop()
                df_subset = np.array(subset_list)
                df_mean = df_subset.mean(axis=1).reshape(-1,1)
                df_cusum = (df_subset - df_mean).cumsum(axis=1)
                r = df_cusum.max(axis=1) - df_cusum.min(axis=1) + np.spacing(1)
                s = df_subset.std(axis=1, ddof=0) + np.spacing(1)
                r_s_dict.append({'R_S': (r / s).mean(), 'N': k})
            log_r_s=[]
            log_n=[]
            for i in range(len(r_s_dict)):
                log_r_s.append(np.log(r_s_dict[i]['R_S']))
                log_n.append(np.log(r_s_dict[i]['N']))
            try:
                res = np.polyfit(log_n, log_r_s, 1)[0]
            except:
                res = None
            return res

本章节中，我们基于 Python 实现了相同的因子计算并统计计算时间。

通过引入了 joblib 库中的 Parallel 方法，通过设置其 n_jobs 参数模拟不同 CPU 核数的运行环境。

在性能测试中，使用 Python 定义测算基金日收益率的 getLog() 函数和执行因子计算任务的 main() 函数 (因子计算通过 numpy, pandas, scipy 等库实现)：

def getLog(value):
    diff_value = np.diff(value)
    rolling_value = np.roll(value, 1)
    rolling_value = np.delete(rolling_value, [0])
    return np.insert(np.true_divide(diff_value, rolling_value), 0, np.nan)

def main(li):
    value = np.array(li["value"])
    price = np.array(li["price"])
    log = np.array(li["log"])
    getAnnualReturn(value)
    getAnnualVolatility(value)
    getAnnualSkew(value)
    getAnnualKur(value)
    getSharp(value)
    getMaxDrawdown(value)
    getDrawdownRatio(value)
    getBeta(value, price)
    getAlpha(value, price)
    calHurst(log, 2)

使用 Python API 进行取数等数据处理操作，使用Python对其进行数据对齐和计算基金日收益率的操作。统计整个计算流程的时间：

s = ddb.session()
s.connect("127.0.0.1", 8848, "admin", "123456")
start = time.time()
fund_OLAP = s.loadTable(dbPath="dfs://fund_OLAP", tableName="fund_OLAP").select("*").toDF().sort_values(['tradingDate'])
fund_hs_OLAP = s.loadTable(dbPath="dfs://fund_OLAP", tableName="fund_hs_OLAP").select("*").toDF()
fund_hs_OLAP.rename(columns={'tradingDate': 'hstradingDate'}, inplace=True)
fund_hs_OLAP = fund_hs_OLAP.sort_values(['hstradingDate'])
fund_dui_OLAP = pd.merge_asof(fund_OLAP, fund_hs_OLAP, left_on="tradingDate", right_on="hstradingDate").sort_values(['fundNum_x', 'tradingDate'])
fund_dui_OLAP = fund_dui_OLAP[fund_dui_OLAP['tradingDate'] == fund_dui_OLAP['hstradingDate']]
fund_dui_OLAP.reset_index(drop=True, inplace=True)
fund_dui_OLAP.drop(columns=['fundNum_y', 'hstradingDate'], inplace=True)
fund_dui_OLAP.columns = ['tradingDate', 'fundNum', 'value', 'price']
fund_dui_OLAP["log"] = pd.Series(getLog(fund_dui_OLAP["value"]))
list = fund_dui_OLAP[(fund_dui_OLAP['tradingDate'] >= datetime(2018, 5, 24)) & (fund_dui_OLAP['tradingDate'] <= datetime(2021, 5, 27))].groupby('fundNum')
Parallel(n_jobs=1)(delayed(main)(i) for _,i in list)
end = time.time()
print(end-start)

• 单任务（单位：秒）

  CPU数	DolphinDB 响应时间	Python 响应时间	Python/DolphinDB
  1	26.03	112.34	4.32
  6	7.23	55.35	7.66
  12	4.23	36.19	8.56

• 多任务（此处为5用户，单位：秒）

  CPU数	DolphinDB 响应时间	Python 响应时间	Python/DolphinDB
  1	140.49	511.89	3.64
  6	27.52	226.45	8.22
  12	15.66	224.80	14.35

注意：教程中我们只选取了部分测试结果进行展示。

在控制其它变量一致的前提下，无论是单任务还是多任务，在不同 CPU 核数环境下 DolphinDB 均表现出了比 Python 更优越的性能。其中性能差异最高可接近 Python 的14倍，平均性能超10倍。究其原因，主要有以下几点：

• DolphinDB 强大的向量化计算能力：DolphinDB 最大程度上兼顾了向量化计算的性能，即其计算引擎的设计针对涉及到向量的计算进行了性能上的优化，而 Python 作为解释型脚本语句，每运行一句都要进行相应的解释，且对向量化计算能力的优化较 DolphinDB 有一定的欠缺，这使得针对数组的向量化计算显示出了比 Python 更优越的性能；
• DolphinDB 丰富的内置函数：DolphinDB 预定义了1000多个可以直接调用的函数。相较于 Python 减少了因子定义的代码量和封装次数，在计算过程中性能损耗更少；
• DolphinDB 自带的持久化数据存储：DolphinDB 内置数据存储引擎，可以将数据按照不同分区方式存入分布式数据库表，因子计算时的数据读取更高效；Python 在因子计算的过程中需要引入外部数据源，数据读取效率相对较低。

本教程基于3000多只基金的日净值数据，对比测试了不同 CPU 核数环境下以及任务提交数下，Python 和 DolphinDB 计算相同因子的性能差异。

可以看到，由于DolphinDB具有强大的向量化计算的能力，包含丰富的内置函数功能，以及具备自带的持久化数据存储，因而无论在多线程还是多任务情况下，DolphinDB 在计算因子时都有相较于 Python 更为优异的表现。因此相较之下，DolphinDB 表现出更为显著的优势。

• 测试数据
• 数据导入到 DolphinDB 的脚本
• 基于 DolphinDB 的基金日频因子实现与性能测试
• 基于 Python 的十个基金日频因子实现与性能测试