如何用自适应方法替换FVC计算函数中的常量阈值？

问题 #

我写了一个计算用力肺活量（FVC）的函数，逻辑是检测数据序列中负值出现前的最后一个正值，代码如下：

def FVC(self, data):
    count = 0
    for i in range(len(data)):
        # 检测是否存在足够多的负值以判断吸气阶段开始
        if (data[i][1] > 1) and (data[i][2] < 0):
            count += 1
        elif count > 110:
            return data[i-115][1]
        else:
            count = 0

目标序列示例 #

目标值为[8340.0, 3.0616, 0.0]：

[8280.0, 3.0611, 0.01]
[8290.0, 3.0612, 0.01]
[8300.0, 3.0613, 0.01]
[8310.0, 3.0614, 0.01]
[8320.0, 3.0615, 0.01]
[8330.0, 3.0616, 0.01]
[8340.0, 3.0616, 0.0]
[8350.0, 3.0615, -0.01125]
[8360.0, 3.0614, -0.01125]
[8370.0, 3.0613, -0.01125]
[8380.0, 3.0611, -0.01625]
[8390.0, 3.0609, -0.0275]
[8400.0, 3.0605, -0.033125]
[8410.0, 3.0601, -0.044375]
[8420.0, 3.0595, -0.055625]
[8430.0, 3.0589, -0.06625]
[8440.0, 3.0582, -0.06625]
[8450.0, 3.0575, -0.06625]
[8460.0, 3.0568, -0.0775]
[8470.0, 3.056, -0.0775]
[8480.0, 3.0551, -0.08875]
[8490.0, 3.0541, -0.105]
[8500.0, 3.0527, -0.1325]
[8510.0, 3.051, -0.17125]
[8520.0, 3.0489, -0.21]
[8530.0, 3.0465, -0.24375]
[8540.0, 3.0437, -0.2825]
[8550.0, 3.0406, -0.31]
[8560.0, 3.0371, -0.34312]
[8570.0, 3.0334, -0.37062]
[8580.0, 3.0294, -0.39875]
[8590.0, 3.0251, -0.43125]
[8600.0, 3.0204, -0.47625]
[8610.0, 3.0151, -0.52625]
[8620.0, 3.0091, -0.5975]

异常数据示例 #

数据中会出现短暂负值后又回到正值的情况，导致误判：

[6860.0, 3.0468, 0.0]
[6870.0, 3.0467, -0.005625]
[6880.0, 3.0466, -0.01125]
[6890.0, 3.0465, -0.01125]
[6900.0, 3.0464, -0.01125]
[6910.0, 3.0463, -0.01125]
[6920.0, 3.0461, -0.01125]
[6930.0, 3.046, -0.01125]
[6940.0, 3.0459, -0.01125]
[6950.0, 3.0457, -0.021875]
[6960.0, 3.0455, -0.021875]
[6970.0, 3.0452, -0.0275]
[6980.0, 3.0449, -0.033125]
[6990.0, 3.0445, -0.033125]
[7000.0, 3.0443, -0.021875]
[7010.0, 3.0441, -0.021875]
[7020.0, 3.0439, -0.021875]
[7030.0, 3.0437, -0.021875]
[7040.0, 3.0435, -0.021875]
[7050.0, 3.0433, -0.01125]
[7060.0, 3.0433, -0.005625]
[7070.0, 3.0433, 0.0]
[7080.0, 3.0433, 0.0]
[7090.0, 3.0433, 0.0]
[7100.0, 3.0433, 0.0]
[7110.0, 3.0433, 0.0]
[7120.0, 3.0433, 0.0]
[7130.0, 3.0433, 0.0]
[7140.0, 3.0433, 0.0]
[7150.0, 3.0433, 0.0]
[7160.0, 3.0434, 0.01]
[7170.0, 3.0435, 0.01]

现在我用固定的110次连续负值来判断真正的吸气阶段，但这个常量适应性差，想换成自适应的判断逻辑，怎么实现？

解决方案 #

1. 基于趋势的自适应判断 #

真正的吸气阶段（FVC之后），负值的绝对值会呈现持续增大的趋势，而异常波动的负值会先小幅度变化后回归。可以通过跟踪连续负值的变化特征判断：

• 记录连续负值序列，计算其绝对值的变化斜率
• 当斜率持续为正（绝对值越来越大）且连续次数超过动态阈值时，判定为真正的吸气阶段

示例代码：

def FVC(self, data):
    neg_sequence = []
    last_positive_idx = 0
    for i in range(len(data)):
        val = data[i][2]
        if data[i][1] > 1:
            if val < 0:
                neg_sequence.append(val)
                # 用最近20个负值判断趋势
                if len(neg_sequence) >= 20:
                    abs_vals = [abs(x) for x in neg_sequence[-20:]]
                    slope = (abs_vals[-1] - abs_vals[0]) / 20
                    # 斜率大于最小变化率，且连续次数超过数据总长度的1/20
                    if slope > 0.005 and len(neg_sequence) > len(data) // 20:
                        return data[last_positive_idx][1]
            else:
                last_positive_idx = i
                neg_sequence = []
    return data[last_positive_idx][1]

2. 状态机阶段划分 #

把呼吸过程划分为呼气阶段、疑似吸气阶段、确认吸气阶段三个状态，通过状态转移实现自适应判断：

• 呼气阶段：持续记录最后一个正值的位置，遇到负值进入疑似状态
• 疑似吸气阶段：统计连续负值的数量和变化，若回归正值则回到呼气阶段；若负值持续且绝对值增大，则进入确认状态
• 确认吸气阶段：返回之前记录的最后一个正值

3. 统计特征动态阈值 #

先遍历数据统计呼气阶段的特征，再生成动态阈值：

• 统计呼气阶段（FVC之前）连续非负值的平均长度
• 将判断阈值设为该平均长度的1/3~1/4，过滤短暂异常波动

示例代码：

def FVC(self, data):
    # 先统计呼气阶段的连续非负值长度
    expiratory_lengths = []
    current_len = 0
    for i in range(len(data)):
        val = data[i][2]
        if val >= 0 and data[i][1] > 1:
            current_len += 1
        else:
            if current_len > 0:
                expiratory_lengths.append(current_len)
                current_len = 0
    # 生成自适应阈值
    avg_exp_len = sum(expiratory_lengths)/len(expiratory_lengths) if expiratory_lengths else 50
    adaptive_threshold = int(avg_exp_len * 0.2)

    # 正式检测FVC
    count = 0
    last_positive_idx = 0
    for i in range(len(data)):
        if (data[i][1] > 1) and (data[i][2] < 0):
            count += 1
            if count >= adaptive_threshold:
                # 额外验证后续5个点的趋势
                if all(data[j][2]<0 and abs(data[j][2])>abs(data[j-1][2]) for j in range(max(i-5,0), i+1)):
                    return data[last_positive_idx][1]
        else:
            if data[i][1]>1 and data[i][2]>=0:
                last_positive_idx = i
            count = 0
    return data[last_positive_idx][1]

关键注意点 #

• 保留data[i][1]>1的有效性判断，避免无效数据干扰
• 加入滞后验证：达到阈值后再验证后续数据点的趋势，减少误判
• 可根据数据集调整斜率阈值、比例系数等参数，或引入简单分类模型区分正常吸气和异常波动

内容的提问来源于stack exchange，提问作者Noerigarnhy