refactor: 提取 StatCollector 统计学习组件

将 ExpBase::cron_proc() 中的 DAA::STA 统计学习逻辑提取到独立的
StatCollector 工具类，统一管理分布统计的生命周期（样本累积、分布
初始化、DB2 持久化、置信区间更新）。同时将 exec_task() 与
task_mon_pro() 中的任务相关 STA 操作也委托给 StatCollector。

新增:
- eqpalg/utility/stat_collector.h  -- StatCollector 接口
- eqpalg/utility/stat_collector.cpp -- StatCollector 实现

修改:
- eqpalg/algs/exp_base.h  -- 替换 sta_ptr_ 为 stat_collector_
- eqpalg/algs/exp_base.cpp -- cron_proc/reload_ci_dist/reset_dev_data/
                               exec_task/task_mon_pro 委托给 StatCollector

eqpalg/algs/exp_base.cpp

@@ -49,6 +49,7 @@ int ExpBase::init() {
       ret += this->reload_config_up_down(); /*7.上下限*/
       this->reload_ci_dist();
       this->last_load_time_ = chrono::system_clock::now();
+      stat_collector_.configure(rule_id_, rule_name_, dist_mode_, is_learning_);
     }
     if (exp_type_ == ExpType::BoundHoldTime) {
       ret += this->reload_config_up_down_hold_time(); /*8.上下限-保持时间*/
@@ -169,10 +170,7 @@ std::vector<AlarmInfo> ExpBase::exec_task(mix_cc::time_range_t time_range) {
           << mix_cc::mix_time_t(time_range.get_right()).to_formatted_time()
           << endl;
       std::string sample_id = this->get_id(time_range);
-      if (sta_ptr_ == nullptr) {
+      stat_collector_.ensureInitialized();
-        logger_->Debug() << rule_name_ << " sta_ptr_ == nullptr" << endl;
-        sta_ptr_ = std::make_unique<DAA::STA>(rule_id_, rule_name_);
-      }
 
       logger_->Debug() << "reset_data()------1" << endl;
       if ((time_range.get_right() - time_range.get_left()) <
@@ -192,7 +190,7 @@ std::vector<AlarmInfo> ExpBase::exec_task(mix_cc::time_range_t time_range) {
         }
       }
       logger_->Debug() << "reset_data()------1" << endl;
-      this->sta_ptr_->reset_data(); /*参数重置*/
+      stat_collector_.resetData(); /*参数重置*/
       logger_->Debug() << "reset_data()------2" << endl;
       logger_->Debug() << "task_seq:" << task_seq << std::endl;
       auto &data_record = TaskShm::TaskRecordPtr.get()
@@ -202,12 +200,12 @@ std::vector<AlarmInfo> ExpBase::exec_task(mix_cc::time_range_t time_range) {
       logger_->Debug() << "dataSize:" << data_record.size() << std::endl;
       for (size_t j = 0; j < data_record.size(); j++) {
         double dataJ = data_record[j];
-        this->sta_ptr_->dist_add(dataJ);
+        stat_collector_.distAdd(dataJ);
       }
       logger_->Debug() << "reset_data()------3" << endl;
-      auto store_res = this->sta_ptr_->task_store_db2(sample_id);
+      auto store_res = stat_collector_.taskStoreDb2(sample_id);
       logger_->Debug() << "reset_data()------4" << endl;
-      this->sample_result_ = this->sta_ptr_->get_sample_stat_str();
+      this->sample_result_ = stat_collector_.getSampleStatStr();
       this->update_t_sample_mag(store_res);
       this->alarm_poster_.zmqp_send(912, this->sample_result_);
       TaskShm::TaskRecordPtr.get()->erase(exp_type_ * 1000 + task_seq);
@@ -489,50 +487,13 @@ int ExpBase::cron_proc() {
   if (exp_type_ == ExpType::Bound || exp_type_ == ExpType::CondBound ||
       exp_type_ == ExpType::BoundHoldTime) {
     /*只保存有上下限的,2-监控变量-上下限；4-动作反馈-上下限；5-监控变量-上下限-持续*/
-    if (sta_ptr_ == nullptr) {
-      logger_->Debug() << rule_name_ << " sta_ptr_ == nullptr" << endl;
-      sta_ptr_ = std::make_unique<DAA::STA>(rule_id_, rule_name_);
-      this->sta_ptr_->update_ci_dist();
-      last_load_time_ = chrono::system_clock::now();
-    }
-
-    if (now_time_ - last_load_time_ > hours(CronUpdateDelay)) {
-      this->sta_ptr_->update_ci_dist();
-      last_load_time_ = now_time_;
-    }
     this->rule_stat_.stat_values.clear();
     SingletonTemp<EqpStat>::GetInstance().get_stat_values(this->rule_id_,
                                                           this->rule_stat_);
     if (!this->rule_stat_.stat_values.empty()) {
-      size_data = this->rule_stat_.stat_values.size();
+      size_data = stat_collector_.processCron(
-      logger_->Debug() << rule_name_ << ",this->rule_stat_ size:" << size_data
+          this->rule_stat_.stat_values, now_time_,
-                       << endl;
+          CronUpdateDelay, last_load_time_);
-      if (sta_ptr_->is_init()) {
-        for (int i = 0; i < size_data; i++) {
-          sta_ptr_->dist_add(this->rule_stat_.stat_values[i]);
-        }
-      } else {
-        double max_data =
-            *(std::max_element(this->rule_stat_.stat_values.begin(),
-                               this->rule_stat_.stat_values.end()));
-        double min_data =
-            *(std::min_element(this->rule_stat_.stat_values.begin(),
-                               this->rule_stat_.stat_values.end()));
-        double range = (max_data - min_data) / double(DAA::STA_SIZE_MIN);
-        logger_->Debug() << "max:" << max_data << ",min:" << min_data
-                         << ",range:" << range << endl;
-        if (range < 0.1) {
-          range = 0.1;
-        }
-        if (sta_ptr_->init(range, min_data)) {
-          for (int i = 0; i < size_data; i++) {
-            sta_ptr_->dist_add(this->rule_stat_.stat_values[i]);
-          }
-        }
-      }
-      logger_->Debug() << rule_name_ << ",sta_ptr_ size:" << sta_ptr_->size()
-                       << endl;
-      sta_ptr_->store_db2();
     }
     this->rule_stat_.stat_values.clear();
     this->rule_stat_.stat_values.shrink_to_fit();
@@ -803,31 +764,17 @@ int ExpBase::reload_ci_dist() {
       this->exp_type_ != ExpType::BoundHoldTime) {
     return 0;
   }
-  last_load_time_ = now_time_;
+  double newDown, newUp;
-  if (this->dist_mode_ != 0) {
+  if (stat_collector_.reloadCiDist(newDown, newUp, now_time_, last_load_time_)) {
-    mix_cc::float_range_t dist_range;
+    this->limit_down_ = newDown;
-    if (this->dist_mode_ == DistMode::Online) {
+    this->limit_up_ = newUp;
-      dist_range = DAA::STA::select_from_t_rule_feature(this->rule_id_);
+    bound_checker_.setLimits(newDown, newUp);
-    } else if (this->dist_mode_ == DistMode::Offline) {
-      dist_range = DAA::STA::select_from_t_sample_mag(this->rule_id_);
-    }
-
-    if (dist_range.get_distance() == 0 ||
-        dist_range.get_left() > dist_range.get_right()) {
-      logger_->Error() << "区间不合法![" << dist_range.get_left() << ","
-                       << dist_range.get_right() << "]" << endl;
-      return -1;
-    }
-    this->limit_down_ = dist_range.get_left();
-    this->limit_up_ = dist_range.get_right();
-    bound_checker_.setLimits(limit_down_, limit_up_);
     this->rule_stat_.limit_down = limit_down_;
     this->rule_stat_.limit_up = limit_up_;
-    this->rule_stat_.current_value =
+    this->rule_stat_.current_value = (limit_down_ + limit_up_) / 2;
-        (this->limit_down_ + this->limit_up_) / 2;
     logger_->Info()
-        << "更新置信区间，[" << dist_range.get_left() << ","
+        << "更新置信区间，[" << newDown << ","
-        << dist_range.get_right() << "]"
+        << newUp << "]"
         << ",type[0-手动设置的区间;1-在线更新的区间；2-离线分析的区间]:"
         << this->dist_mode_ << endl;
   }
@@ -873,7 +820,7 @@ void ExpBase::task_mon_pro() {
             ->
             operator[](exp_type_ * 1000 + task_seq)
             .data_record.push_back(res.value);
-        this->sta_ptr_->running_stat_add(res.value);
+        stat_collector_.runningStatAdd(res.value);
       }
     }
   }
@@ -1034,12 +981,7 @@ void ExpBase::reset_dev_data() {
   if (glob_process_type == ProcessType::kMon) {
     if (exp_type_ == ExpType::Bound || exp_type_ == ExpType::CondBound ||
         exp_type_ == ExpType::BoundHoldTime) {
-      int res = DAA::STA::delete_statistics_data(this->rule_id_);
+      stat_collector_.reset(rule_id_);
-      if (res != 0) {
-        logger_->Debug() << "DAA::STA::delete_statistics_data(this->rule_id_) "
-                            "存在删除失败的表！"
-                         << endl;
-      }
       this->reload_config_up_down(); /*7.上下限*/
       this->reload_ci_dist();
       this->last_load_time_ = chrono::system_clock::now();

eqpalg/algs/exp_base.h

@@ -22,6 +22,7 @@
 #include <eqpalg/utility/StatExp.hpp>
 #include <eqpalg/utility/item2chinese.hpp>
 #include <eqpalg/utility/bound_checker.h>
+#include <eqpalg/utility/stat_collector.h>
 #include <eqpalg/utility/fb_state_machine.h>
 #include <glob/SingletonTemplate.h>
 #include <iomanip>
@@ -95,7 +96,7 @@ public:
    *   删除
    * 四张表(T_RULE_SAMPLE_1D,T_RULE_SAMPLE_1D_INFO,T_SAMPLE_RECORD，T_SAMPLE_RECORD，T_RULE_SAMPLE_FEATURE)rule_id_对应数据
    * cron----
-   * 重置 sta_ptr_(sta_ptr_.reset())
+   * 重置 stat_collector_(stat_collector_.reset())
    */
   virtual void reset_dev_data() override;
 
@@ -171,7 +172,7 @@ protected:
 
   TimeDur hold_time_ = 0ms;
 
-  std::unique_ptr<DAA::STA> sta_ptr_;
+  StatCollector stat_collector_;
 
   std::map<std::string, std::unique_ptr<HoldTime>>
       hold_times_;

eqpalg/utility/stat_collector.cpp

@@ -0,0 +1,112 @@
+// eqpalg/utility/stat_collector.cpp
+#include <eqpalg/utility/stat_collector.h>
+#include <eqpalg/feature_extraction/daa.h>
+#include <eqpalg/define/public.h>
+#include <algorithm>
+#include <log4cplus/LOG.h>
+
+StatCollector::~StatCollector() = default;
+
+void StatCollector::configure(const std::string& ruleId,
+                               const std::string& ruleName,
+                               int distMode, bool isLearning) {
+    rule_id_ = ruleId;
+    rule_name_ = ruleName;
+    dist_mode_ = distMode;
+    is_learning_ = isLearning;
+    configured_ = true;
+}
+
+int StatCollector::processCron(const std::vector<double>& statValues,
+                                TimePoint now, int cronDelayHours,
+                                TimePoint& lastLoadTime) {
+    if (statValues.empty()) return 0;
+
+    int size_data = static_cast<int>(statValues.size());
+
+    if (sta_ptr_ == nullptr) {
+        sta_ptr_ = std::make_unique<DAA::STA>(rule_id_, rule_name_);
+        sta_ptr_->update_ci_dist();
+        lastLoadTime = now;
+    }
+
+    if (now - lastLoadTime > std::chrono::hours(cronDelayHours)) {
+        sta_ptr_->update_ci_dist();
+        lastLoadTime = now;
+    }
+
+    if (sta_ptr_->is_init()) {
+        for (int i = 0; i < size_data; i++) {
+            sta_ptr_->dist_add(statValues[i]);
+        }
+    } else {
+        double max_data = *std::max_element(statValues.begin(), statValues.end());
+        double min_data = *std::min_element(statValues.begin(), statValues.end());
+        double range = (max_data - min_data) / static_cast<double>(DAA::STA_SIZE_MIN);
+        if (range < 0.1) range = 0.1;
+        if (sta_ptr_->init(range, min_data)) {
+            for (int i = 0; i < size_data; i++) {
+                sta_ptr_->dist_add(statValues[i]);
+            }
+        }
+    }
+
+    sta_ptr_->store_db2();
+    return size_data;
+}
+
+bool StatCollector::reloadCiDist(double& limitDown, double& limitUp,
+                                  TimePoint now, TimePoint& lastLoadTime) {
+    if (dist_mode_ == 0) return false;  // 手动模式，不自动加载
+
+    lastLoadTime = now;
+
+    mix_cc::float_range_t dist_range;
+    if (dist_mode_ == DistMode::Online) {
+        dist_range = DAA::STA::select_from_t_rule_feature(rule_id_);
+    } else if (dist_mode_ == DistMode::Offline) {
+        dist_range = DAA::STA::select_from_t_sample_mag(rule_id_);
+    } else {
+        return false;
+    }
+
+    if (dist_range.get_distance() == 0 ||
+        dist_range.get_left() > dist_range.get_right()) {
+        return false;
+    }
+
+    limitDown = dist_range.get_left();
+    limitUp = dist_range.get_right();
+    return true;
+}
+
+void StatCollector::reset(const std::string& ruleId) {
+    DAA::STA::delete_statistics_data(ruleId);
+    sta_ptr_.reset();
+}
+
+void StatCollector::ensureInitialized() {
+    if (sta_ptr_ == nullptr) {
+        sta_ptr_ = std::make_unique<DAA::STA>(rule_id_, rule_name_);
+    }
+}
+
+void StatCollector::resetData() {
+    sta_ptr_->reset_data();
+}
+
+void StatCollector::distAdd(double value) {
+    sta_ptr_->dist_add(value);
+}
+
+bool StatCollector::taskStoreDb2(const std::string& sampleId) {
+    return sta_ptr_->task_store_db2(sampleId);
+}
+
+std::string StatCollector::getSampleStatStr() {
+    return sta_ptr_->get_sample_stat_str();
+}
+
+void StatCollector::runningStatAdd(double value) {
+    sta_ptr_->running_stat_add(value);
+}

eqpalg/utility/stat_collector.h

@@ -0,0 +1,94 @@
+// eqpalg/utility/stat_collector.h
+#pragma once
+
+#include <chrono>
+#include <memory>
+#include <string>
+#include <vector>
+
+// 前向声明
+namespace DAA { class STA; }
+
+using TimePoint = std::chrono::system_clock::time_point;
+
+/**
+ * @brief 统计学习收集器
+ *
+ * 管理 DAA::STA 分布统计的生命周期：
+ * - 样本累积（dist_add）
+ * - 分布初始化与 DB2 持久化
+ * - 置信区间更新
+ */
+class StatCollector {
+public:
+    StatCollector() = default;
+    ~StatCollector();
+
+    /**
+     * @brief 配置统计收集器
+     * @param ruleId     规则 ID
+     * @param ruleName   规则名称
+     * @param distMode   分布模式（0=手动, 1=在线, 2=离线）
+     * @param isLearning 是否启用自学习
+     */
+    void configure(const std::string& ruleId, const std::string& ruleName,
+                   int distMode, bool isLearning);
+
+    /**
+     * @brief cron 进程：从共享内存获取累积样本，更新分布并持久化
+     * @param statValues  累积的样本值
+     * @param now         当前时间
+     * @param cronDelayHours  cron 更新延迟（小时）
+     * @param lastLoadTime [in/out] 上次加载时间
+     * @return 处理的样本数量
+     */
+    int processCron(const std::vector<double>& statValues,
+                    TimePoint now, int cronDelayHours,
+                    TimePoint& lastLoadTime);
+
+    /**
+     * @brief 加载置信区间到上下限
+     * @param limitDown [out] 新的下限
+     * @param limitUp   [out] 新的上限
+     * @param now       当前时间
+     * @param lastLoadTime [out] 更新时间
+     * @return true 成功加载新区间
+     */
+    bool reloadCiDist(double& limitDown, double& limitUp,
+                      TimePoint now, TimePoint& lastLoadTime);
+
+    /**
+     * @brief 删除统计数据和重置
+     */
+    void reset(const std::string& ruleId);
+
+    bool isConfigured() const { return configured_; }
+
+    // ---- task 进程专用接口 ----
+
+    /** @brief 确保 STA 实例已初始化（惰性初始化） */
+    void ensureInitialized();
+
+    /** @brief 重置累积数据，供新 task 使用 */
+    void resetData();
+
+    /** @brief 向分布添加单个值 */
+    void distAdd(double value);
+
+    /** @brief 将 task 结果存储到 DB2 (T_SAMPLE_STAT) */
+    bool taskStoreDb2(const std::string& sampleId);
+
+    /** @brief 获取样本统计字符串，用于 T_SAMPLE_MAG */
+    std::string getSampleStatStr();
+
+    /** @brief 向运行统计中添加值 */
+    void runningStatAdd(double value);
+
+private:
+    std::unique_ptr<DAA::STA> sta_ptr_;
+    std::string rule_id_;
+    std::string rule_name_;
+    int dist_mode_ = 0;
+    bool is_learning_ = true;
+    bool configured_ = false;
+};