聚宽的这些数据处理函数没有对外公开它们的实现细节。不过,你可以查阅一些公共机器学习模块,例如Scikit-Learn或StatsModels,来看看是否有类似的实现。
以下是示例代码:
import numpy as np
from sklearn.preprocessing import StandardScaler
def winsorize_med(x, scale=5, inf2nan=False, axis=0):
# 计算 median 和 mad
med = np.median(x, axis=axis, keepdims=True)
mad = np.median(np.abs(x - med), axis=axis, keepdims=True) / 0.6745
# 进行 winsorization
lower = med - scale * mad
upper = med + scale * mad
if inf2nan:
x[x < lower] = np.nan
x[x > upper] = np.nan
else:
x[x < lower] = lower[x < lower] # 用 lower 值替换小于 lower 的值
x[x > upper] = upper[x > upper] # 用 upper 值替换大于 upper 的值
return x
def neutralize(x, how=None, date=None, axis=0):
# 计算中性化矩阵
weights = x[how].groupby(date).transform(np.linalg.pinv)
# 计算残差
residuals = x - np.dot(weights, x[how])
# 如果对列进行操作,则最后返回的是wide表,需要重新设定列名
# 因为原中性化矩阵索引都是ticker,不是columnName
if axis == 1:
residuals.columns = x.columns
return residuals
def standardlize(x, axis=0):
scaler = StandardScaler()
if axis == 0:
# 将每列数据标准化
x = scaler.fit_transform(x)
elif axis == 1:
# 将每行数据标准化
x = scaler.fit_transform(x.T)
x = x.T
return x
请注意,这里的实现只是一个可能的实现,可能不完全与聚宽的实现相同。因此,你需要对这些函数进行测试来确保它们能够满足你的需求。