flowchart has problem

[taroyutao]

import numpy as np import pandas as pd from adtk.data import validate_series import matplotlib.pyplot as plt from adtk.visualization import plot from adtk.transformer import DoubleRollingAggregate from adtk.detector import ThresholdAD, QuantileAD from adtk.aggregator import AndAggregator

故障检测

from adtk.pipe import Pipenet

steps = { # 长窗口历史数据扫描，规避持续异常跳变 "more_step_roll_aggr": { "model": DoubleRollingAggregate( agg="max", window=(20, 1), diff="diff" ), "input": "original", }, # 确保递增是一直持续存在 "abs_more_step_roll_aggr": { "model": ThresholdAD( high=0 ), "input": "more_step_roll_aggr" },

# 短窗口数据扫描，相对变化率(斜率)判断
"one_step_change": {
    "model": DoubleRollingAggregate(
        agg="mean",
        window=(3, 1),
        diff="rel_diff"
    ),
    "input": "original",
},
"abs_one_step_change": {
    "model": ThresholdAD(high=0.2),
    "input": "one_step_change"
},

"base_level_detect1": {
    "model": ThresholdAD(high=100),
    "input": "original"
},

"positive_level_shift": {
    "model": AndAggregator(),
    "input": ["abs_more_step_roll_aggr", "abs_one_step_change", "base_level_detect"]
}

}

pipenet = Pipenet(steps) pipenet.plot_flowchart()

plt.show()

plt.savefig('net1.png')

[ljluestc]

import numpy as np
import pandas as pd
from adtk.data import validate_series
import matplotlib.pyplot as plt
from adtk.visualization import plot
from adtk.transformer import DoubleRollingAggregate
from adtk.detector import ThresholdAD, QuantileAD
from adtk.aggregator import AndAggregator
from adtk.pipe import Pipenet

# Example data
np.random.seed(0)
data = pd.Series(np.random.randn(100))

# Validate the series
data = validate_series(data)

# Define the pipeline steps
steps = {
    # Long window historical data scan
    "more_step_roll_aggr": {
        "model": DoubleRollingAggregate(
            agg="max",
            window=(20, 1),
            diff="diff"
        ),
        "input": "original",
    },
    # Ensure increment is consistently present
    "abs_more_step_roll_aggr": {
        "model": ThresholdAD(
            high=0
        ),
        "input": "more_step_roll_aggr",
    },

    # Short window data scan
    "one_step_change": {
        "model": DoubleRollingAggregate(
            agg="mean",
            window=(3, 1),
            diff="rel_diff"
        ),
        "input": "original",
    },
    "abs_one_step_change": {
        "model": ThresholdAD(high=0.2),
        "input": "one_step_change",
    },

    "base_level_detect1": {
        "model": ThresholdAD(high=100),
        "input": "original"
    },

    "positive_level_shift": {
        "model": AndAggregator(),
        "input": ["abs_more_step_roll_aggr", "abs_one_step_change", "base_level_detect1"]
    }
}

# Initialize Pipenet with the steps
pipenet = Pipenet(steps)

# Plot the flowchart
pipenet.plot_flowchart()

# Save the figure
plt.savefig('net1.png')
plt.show()