如何开发算法检测流量监测折线的近似相邻与同向性？

核心检测条件 #

需同时满足两个核心要求：

1. 两条折线均穿过零点（或流量值接近零点）
2. 两条折线同向且角度几乎一致（斜率符号相同，且斜率相对误差在允许范围内）

算法步骤 #

1. 数据准备

假设每5分钟同步采集两条折线的流量数据，存储为两个时间对齐的点集：

• 负流量折线（红色）：points_neg = [(t₁, v₁), (t₂, v₂), ..., (tₙ, vₙ)]，其中t为时间戳，v为流量值
• 正流量折线（绿色）：points_pos = [(t₁, u₁), (t₂, u₂), ..., (tₙ, uₙ)]

2. 零点穿越检测

检查每条折线是否满足任一条件：

• 相邻两个数据点的流量值符号相反（vᵢ₋₁ * vᵢ < 0），说明折线穿过零点
• 最新数据点的流量值绝对值小于设定的零点阈值（|vᵢ| < ε），认为折线接近零点

3. 斜率（角度）计算与一致性判断

由于时间间隔固定为5分钟，可简化斜率计算为相邻点的流量变化量：

• 对每条折线，取最近W个数据点（建议3-5个），计算相邻点的斜率，再求平均值得到平均斜率
• 同向判断：两条折线的平均斜率符号必须相同（同时为正或同时为负）
• 角度一致判断：
  • 若两条折线的平均斜率绝对值均小于水平阈值（|avg_slope| < θ），判定为近似水平，符合条件
  • 否则计算两个平均斜率的相对误差，若误差小于设定的斜率阈值（δ），则判定角度几乎一致

伪代码实现 #

# 可根据实际数据调整的阈值
EPS_ZERO = 0.1       # 流量接近零点的允许误差
DELTA_SLOPE = 0.1    # 斜率相对误差的允许阈值
THETA_FLAT = 0.05    # 判定折线为水平的斜率阈值
WINDOW_SIZE = 3      # 计算平均斜率的滑动窗口大小（最近3组数据）

def check_approx_adjacent(points_neg, points_pos):
    # 检测负流量折线是否穿过/接近零点
    neg_cross = False
    for i in range(1, len(points_neg)):
        prev_v, curr_v = points_neg[i-1][1], points_neg[i][1]
        if (prev_v * curr_v < 0) or (abs(curr_v) < EPS_ZERO):
            neg_cross = True
            break
    if not neg_cross:
        return False

    # 检测正流量折线是否穿过/接近零点
    pos_cross = False
    for i in range(1, len(points_pos)):
        prev_u, curr_u = points_pos[i-1][1], points_pos[i][1]
        if (prev_u * curr_u < 0) or (abs(curr_u) < EPS_ZERO):
            pos_cross = True
            break
    if not pos_cross:
        return False

    # 计算折线的平均斜率
    def get_avg_slope(points):
        if len(points) < WINDOW_SIZE + 1:
            return None  # 数据量不足，无法计算
        slopes = []
        # 取最近WINDOW_SIZE个斜率
        start_idx = len(points) - WINDOW_SIZE
        for i in range(start_idx, len(points)):
            slope = points[i][1] - points[i-1][1]
            slopes.append(slope)
        return sum(slopes) / len(slopes)

    avg_slope_neg = get_avg_slope(points_neg)
    avg_slope_pos = get_avg_slope(points_pos)
    if avg_slope_neg is None or avg_slope_pos is None:
        return False

    # 判断同向
    if (avg_slope_neg > 0 and avg_slope_pos < 0) or (avg_slope_neg < 0 and avg_slope_pos > 0):
        return False

    # 判断角度一致
    if abs(avg_slope_neg) < THETA_FLAT and abs(avg_slope_pos) < THETA_FLAT:
        return True
    # 计算相对误差
    max_slope = max(abs(avg_slope_neg), abs(avg_slope_pos))
    if max_slope == 0:
        return True
    relative_error = abs(avg_slope_neg - avg_slope_pos) / max_slope
    return relative_error < DELTA_SLOPE

优化建议 #

1. 阈值校准：所有阈值（EPS_ZERO、DELTA_SLOPE等）需要根据实际业务数据调整，可通过历史样本测试确定最优值
2. 斜率计算优化：若时间间隔不固定，需将斜率计算为(vᵢ - vᵢ₋₁)/(tᵢ - tᵢ₋₁)；也可对最近N个点做线性回归，得到更稳定的拟合斜率
3. 动态窗口：根据折线的波动程度动态调整滑动窗口大小，波动大时增大窗口以平滑噪声

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