Skip to content

fix(multifactor): make lazy calculation thread-safe #495

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
fasiondog merged 9 commits into from
Aug 3, 2026
Merged

fix(multifactor): make lazy calculation thread-safe #495

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
d58a90e
fix(multifactor): make lazy calculation thread-safe
woleigegg Jul 27, 2026
27c4426
fix(build): isolate Eigen regression TU from unity build
woleigegg Jul 31, 2026
88c3b85
test(multifactor): thread-safe lazy calculation coverage
woleigegg Jul 31, 2026
beaebc5
test(multifactor): fix golden values and add clone concurrency coverage
woleigegg Jul 31, 2026
33b4ecf
test(build): add eigen package to unit-test target
woleigegg Jul 31, 2026
a01b7d0
test(multifactor): define test MF _calculate out of class
woleigegg Jul 31, 2026
9ba6e6f
fix(multifactor): export calculate_style_residuals symbol
woleigegg Jul 31, 2026
6a63376
chore: add Author header to new files
woleigegg Aug 3, 2026
65d58b9
refactor: rename style_regression to PascalCase StyleRegression
woleigegg Aug 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
187 changes: 57 additions & 130 deletions hikyuu_cpp/hikyuu/trade_sys/multifactor/MultiFactorBase.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@
*/

#include <cmath>
#include <Eigen/Dense>
#include "hikyuu/utilities/thread/algorithm.h"
#include "hikyuu/indicator/crt/ALIGN.h"
#include "hikyuu/indicator/crt/KDATA.h"
Expand All @@ -21,6 +20,7 @@
#include "hikyuu/StockManager.h"
#include "MultiFactorBase.h"
#include "industry_neutralize.h"
#include "StyleRegression.h"

namespace hku {

Expand Down Expand Up @@ -142,7 +142,7 @@ void MultiFactorBase::baseCheckParam(const string& name) const {
}

void MultiFactorBase::paramChanged() {
m_calculated = false;
m_calculated.store(false, std::memory_order_relaxed);
}

void MultiFactorBase::_checkData() {
Expand All @@ -165,17 +165,22 @@ void MultiFactorBase::_checkData() {
m_stks.size());
}

void MultiFactorBase::reset() {
_reset();

std::lock_guard<std::mutex> lock(m_mutex);
void MultiFactorBase::clearCalculatedData() {
m_ref_dates = {};
m_stk_map = {};
m_all_factors = {};
m_date_index = {};
m_stk_factor_by_date = {};
m_ic = {};
m_calculated = false;
}

void MultiFactorBase::reset() {
// 全程持锁：避免与正在进行的 calculate 写写交叉。
// 注意：_reset() 为虚函数，自定义实现不得在锁内重入同一实例需要 m_mutex 的方法。
std::lock_guard<std::mutex> lock(m_mutex);
_reset();
clearCalculatedData();
m_calculated.store(false, std::memory_order_release);
}

MultiFactorPtr MultiFactorBase::clone() {
Expand Down Expand Up @@ -212,7 +217,7 @@ MultiFactorPtr MultiFactorBase::clone() {

p->m_special_category = m_special_category;

p->m_calculated = false;
p->m_calculated.store(false, std::memory_order_relaxed);
// 强制重算，不克隆以下缓存，避免非线程安全
// p->m_stk_map = m_stk_map;
// p->m_date_index = m_date_index;
Expand All @@ -227,7 +232,7 @@ MultiFactorPtr MultiFactorBase::clone() {

void MultiFactorBase::setQuery(const KQuery& query) {
m_query = query;
m_calculated = false;
m_calculated.store(false, std::memory_order_relaxed);
}

void MultiFactorBase::setRefStock(const Stock& stk) {
Expand All @@ -236,7 +241,7 @@ void MultiFactorBase::setRefStock(const Stock& stk) {
HKU_CHECK(ref_dates.size() >= 2, "The dates len is insufficient! current len: {}",
ref_dates.size());
m_ref_stk = tmp_stk;
m_calculated = false;
m_calculated.store(false, std::memory_order_relaxed);
}

void MultiFactorBase::setStockList(const StockList& stks) {
Expand All @@ -246,18 +251,18 @@ void MultiFactorBase::setStockList(const StockList& stks) {
}

m_stks = stks;
m_calculated = false;
m_calculated.store(false, std::memory_order_relaxed);
}

void MultiFactorBase::setRefFactorSet(const FactorSet& factorset) {
HKU_CHECK(!factorset.isNull() && !factorset.empty(), "Input factor set is null or empty!");
m_factorset = factorset;
m_calculated = false;
m_calculated.store(false, std::memory_order_relaxed);
}

void MultiFactorBase::setNormalize(NormPtr norm) {
m_norm = norm;
m_calculated = false;
m_calculated.store(false, std::memory_order_relaxed);
}

void MultiFactorBase::addSpecialNormalize(const string& name, NormalizePtr norm,
Expand Down Expand Up @@ -295,29 +300,25 @@ void MultiFactorBase::addSpecialNormalize(const string& name, NormalizePtr norm,
m_special_style_inds[found_name] = style_inds;
}

m_calculated = false;
m_calculated.store(false, std::memory_order_relaxed);
}

const DatetimeList& MultiFactorBase::getDatetimeList() {
if (!m_calculated) {
calculate();
}
calculate();
return m_ref_dates;
}

const Indicator& MultiFactorBase::getFactor(const Stock& stk) {
HKU_CHECK(getParam<bool>("save_all_factors"),
"param \"save_all_factors\" is false, can't get all factors!");
if (!m_calculated) {
calculate();
}
calculate();
const auto iter = m_stk_map.find(stk);
HKU_CHECK(iter != m_stk_map.cend(), "Could not find this stock: {}", stk);
return m_all_factors[iter->second];
}

const IndicatorList& MultiFactorBase::getAllFactors() {
if (getParam<bool>("save_all_factors") && !m_calculated) {
if (getParam<bool>("save_all_factors")) {
calculate();
} else {
HKU_WARN("param \"save_all_factors\" is false, can't get all factors!");
Expand All @@ -326,9 +327,7 @@ const IndicatorList& MultiFactorBase::getAllFactors() {
}

ScoreRecordList MultiFactorBase::getScores(const Datetime& d) {
if (!m_calculated) {
calculate();
}
calculate();
ScoreRecordList ret;
const auto iter = m_date_index.find(d);
HKU_IF_RETURN(iter == m_date_index.cend(), ret);
Expand Down Expand Up @@ -435,9 +434,7 @@ ScoreRecordList MultiFactorBase::getScores(const Datetime& date, size_t start, s
}

const vector<ScoreRecordList>& MultiFactorBase::getAllScores() {
if (!m_calculated) {
calculate();
}
calculate();
return m_stk_factor_by_date;
}

Expand All @@ -446,9 +443,7 @@ Indicator MultiFactorBase::getIC(int ndays) {
htr("mf param \"save_all_factors\" is false, can't get all factors!, please "
"set it to true if you want to get IC/ICIR!"));

if (!m_calculated) {
calculate();
}
calculate();

std::lock_guard<std::mutex> lock(m_mutex);

Expand Down Expand Up @@ -531,87 +526,8 @@ IndicatorList MultiFactorBase::_getAllReturns(int ndays) const {

// 行业中性化（按行业分组去组内均值）的纯函数实现见 industry_neutralize.h，
// 提取为内部 inline header 供白盒单元测试直接包含调用。

// 计算多元回归中性化后的因子，y为因子，x为多个解释变量（包含常数项）- Eigen版本
static PriceList calculate_residuals(const PriceList& y, const std::vector<PriceList>& x) {
HKU_ASSERT(!x.empty());
size_t n = y.size();
for (const auto& xi : x) {
HKU_ASSERT(xi.size() == n);
}

PriceList residuals(n, Null<price_t>());
size_t k = x.size(); // 解释变量个数

// 构建设计矩阵和因变量向量
Eigen::MatrixXd Xmat(n, k + 1);
Eigen::VectorXd Yvec(n);

// 填充数据 - 第一列为常数项（全1）
Xmat.col(0).setConstant(1.0);

// 标记有效数据点
std::vector<bool> valid(n, true);

for (size_t i = 0; i < n; ++i) {
Yvec(i) = y[i];

// 检查因变量是否有效
if (std::isnan(y[i]) || std::isinf(y[i])) {
valid[i] = false;
continue;
}

// 填充自变量并检查有效性
for (size_t j = 0; j < k; ++j) {
Xmat(i, j + 1) = x[j][i];
if (std::isnan(x[j][i]) || std::isinf(x[j][i])) {
valid[i] = false;
break;
}
}
}

// 计算有效数据点数量
size_t valid_count = std::count(valid.begin(), valid.end(), true);

// 数据点不足
if (valid_count <= k + 1) {
return residuals;
}

// 创建有效数据的子矩阵
Eigen::MatrixXd X_valid(valid_count, k + 1);
Eigen::VectorXd Y_valid(valid_count);

size_t valid_idx = 0;
for (size_t i = 0; i < n; ++i) {
if (valid[i]) {
X_valid.row(valid_idx) = Xmat.row(i);
Y_valid(valid_idx) = Yvec(i);
valid_idx++;
}
}

// 使用QR分解求解线性回归 β = (X'X)^(-1)X'Y
Eigen::VectorXd beta = X_valid.colPivHouseholderQr().solve(Y_valid);

// 检查解是否有效
if (beta.hasNaN()) {
return residuals;
}

// 计算拟合值和残差
Eigen::VectorXd fitted = Xmat * beta;

for (size_t i = 0; i < n; ++i) {
if (valid[i]) {
residuals[i] = y[i] - fitted(i);
}
}

return residuals;
}
// 风格因子中性化残差回归实现见 StyleRegression.cpp，从本类中提取为串行内核，
// 不再在运行时修改进程级 Eigen 线程配置。

vector<IndicatorList> MultiFactorBase::getAllSrcFactors() {
vector<IndicatorList> all_stk_inds;
Expand Down Expand Up @@ -675,9 +591,9 @@ vector<IndicatorList> MultiFactorBase::getAllSrcFactors() {

// 时间截面标准化/归一化
if (m_norm || !m_special_category.empty() || !m_special_style_inds.empty()) {
// 压制 Eigen 内部 OpenMP 并行，避免与外层按日线程池叠加导致线程超载；
// calculate_residuals 内的 Eigen 矩阵均为栈局部对象，外层按日并行天然可重入。
Eigen::setNbThreads(1);
// 风格因子中性化残差回归已提取为串行内核（StyleRegression.cpp），
// 不再在运行时修改进程级 Eigen::setNbThreads，避免并发 MF 互相污染全局配置；
// 外层按日并行天然可重入，回归内部均为栈局部对象。
unordered_map<string, std::pair<PriceList, size_t>> ind_dummy_dict = _buildDummyIndex();
global_parallel_for_index_void(
0, days_total,
Expand Down Expand Up @@ -735,7 +651,7 @@ vector<IndicatorList> MultiFactorBase::getAllSrcFactors() {
style_value[si] = per_factor[j][si][di];
}
}
new_value = calculate_residuals(new_value, style_value_day);
new_value = calculate_style_residuals(new_value, style_value_day);
}

for (size_t si = 0; si < stk_count; si++) {
Expand All @@ -744,9 +660,6 @@ vector<IndicatorList> MultiFactorBase::getAllSrcFactors() {
}
}
});

// 恢复 Eigen 线程数
Eigen::setNbThreads(std::thread::hardware_concurrency());
}

return all_stk_inds;
Expand Down Expand Up @@ -814,12 +727,24 @@ void MultiFactorBase::_buildIndex() {
}

void MultiFactorBase::calculate() {
HKU_IF_RETURN(m_calculated, void());
// Fast path: lock-free acquire 检查是否已 Ready
if (m_calculated.load(std::memory_order_acquire)) {
return;
}

std::lock_guard<std::mutex> lock(m_mutex);
_checkData();

// 锁内二次检查：mutex 已提供慢路径同步，relaxed 即可
if (m_calculated.load(std::memory_order_relaxed)) {
return;
}

// 构建前清理旧结果，确保重试基于干净状态
clearCalculatedData();

try {
_checkData();

{ // 获取所有证券所有对齐后的原始因子
vector<IndicatorList> all_stk_inds = getAllSrcFactors();

Expand All @@ -839,19 +764,21 @@ void MultiFactorBase::calculate() {

// 计算完成后创建截面索引
_buildIndex();
} catch (const std::exception& e) {
HKU_ERROR(e.what());
} catch (...) {
HKU_ERROR_UNKNOWN;
}

if (!getParam<bool>("save_all_factors")) {
m_all_factors = {};
m_stk_map = {};
if (!getParam<bool>("save_all_factors")) {
m_all_factors = {};
m_stk_map = {};
}
} catch (...) {
// 失败清理：所有异步子任务已在 wait_and_drain 语义下结束，
// 清除基类半成品，保持未计算状态，原异常向上传播，允许下一调用者重试。
clearCalculatedData();
m_calculated.store(false, std::memory_order_relaxed);
throw;
}

// 更新计算状态
m_calculated = true;
// Publish：release 保证此前所有写入对后续 acquire 读取可见
m_calculated.store(true, std::memory_order_release);
}

} // namespace hku
Loading
Loading