mcpField
T
time-series-decomposer

time-series-decomposer

github.com/dkyazzentwatwa数据与数据库12459 次安装29 周前更新
代码仓库

关于 time-series-decomposer

time-series-decomposer 是用于分析时间序列的 Python 工具，可拆分趋势、季节性和残差。它还提供异常分析、可视化和基础预测，适合处理周期数据的分析工作。

主要功能

  1. 01加法与乘法分解
  2. 02趋势与季节性分析
  3. 03残差异常检测
  4. 04组件图及 ACF/PACF 图
  5. 05趋势与季节性预测

适用场景

  1. 01分析销售数据的趋势与季节性
  2. 02标记每日指标中的异常值
  3. 03基于季节性数据预测后续周期

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/dkyazzentwatwa/chatgpt-skills --skill time-series-decomposer

安装后在 Claude Code 中输入/time-series-decomposer确认已加载。

附带文件scripts/requirements.txtscripts/ts_decomposer.py

详细介绍

时间序列分解器

从时间序列数据中提取趋势、季节性和残差成分，并提供可视化和基础预测功能。

功能

  • 分解：加法模型和乘法模型
  • 趋势提取：移动平均、多项式拟合
  • 季节性检测：自动检测并提取周期性模式
  • 残差分析：识别残差中的异常
  • 可视化：成分图、ACF/PACF
  • 基础预测：趋势外推、季节性朴素预测

快速开始

from ts_decomposer import TimeSeriesDecomposer

decomposer = TimeSeriesDecomposer()
decomposer.load_csv("sales.csv", date_col="date", value_col="revenue")

# Decompose
result = decomposer.decompose(period=12)  # Monthly seasonality

print(f"Trend strength: {result['trend_strength']:.2f}")
print(f"Seasonal strength: {result['seasonal_strength']:.2f}")

# Plot components
decomposer.plot_components("decomposition.png")

CLI 用法

# Basic decomposition
python ts_decomposer.py --input data.csv --date date --value sales --period 12

# Multiplicative model
python ts_decomposer.py --input data.csv --date date --value sales --period 12 --model multiplicative

# With forecast
python ts_decomposer.py --input data.csv --date date --value sales --period 12 --forecast 6

# Auto-detect period
python ts_decomposer.py --input data.csv --date date --value sales --auto-period

# Generate plots
python ts_decomposer.py --input data.csv --date date --value sales --period 12 --plot components.png

# Output JSON
python ts_decomposer.py --input data.csv --date date --value sales --period 12 --json

API 参考

TimeSeriesDecomposer 类
class TimeSeriesDecomposer:
    def __init__(self)

    # Data loading
    def load_csv(self, filepath: str, date_col: str, value_col: str,
                date_format: str = None) -> 'TimeSeriesDecomposer'
    def load_series(self, series: pd.Series) -> 'TimeSeriesDecomposer'
    def load_dataframe(self, df: pd.DataFrame, date_col: str,
                      value_col: str) -> 'TimeSeriesDecomposer'

    # Decomposition
    def decompose(self, period: int = None, model: str = "additive") -> dict
    def detect_period(self) -> int
    def extract_trend(self, method: str = "moving_average",
                     window: int = None) -> pd.Series
    def extract_seasonal(self, period: int) -> pd.Series

    # Analysis
    def analyze_trend(self) -> dict
    def analyze_seasonality(self) -> dict
    def analyze_residuals(self) -> dict
    def detect_anomalies(self, threshold: float = 2.0) -> pd.DataFrame

    # Forecasting
    def forecast(self, periods: int, method: str = "trend") -> pd.DataFrame

    # Visualization
    def plot_components(self, output: str) -> str
    def plot_acf_pacf(self, output: str, lags: int = 40) -> str
    def plot_seasonal(self, output: str) -> str

    # Export
    def to_dataframe(self) -> pd.DataFrame
    def summary(self) -> str

分解模型

加法模型

Y(t) = Trend(t) + Seasonal(t) + Residual(t)

适用于季节性变化大致恒定的情况。

result = decomposer.decompose(period=12, model="additive")
乘法模型

Y(t) = Trend(t) * Seasonal(t) * Residual(t)

适用于季节性变化随序列水平而缩放的情况。

result = decomposer.decompose(period=12, model="multiplicative")

输出格式

分解结果
{
    "model": "additive",
    "period": 12,
    "trend_strength": 0.85,      # 0-1, higher = stronger trend
    "seasonal_strength": 0.72,   # 0-1, higher = stronger seasonality
    "components": {
        "observed": [...],       # Original values
        "trend": [...],          # Trend component
        "seasonal": [...],       # Seasonal component
        "residual": [...]        # Residual component
    },
    "seasonal_pattern": {        # Average seasonal effect by period
        1: 0.12,
        2: -0.05,
        ...
    },
    "statistics": {
        "trend_slope": 0.023,
        "trend_r_squared": 0.91,
        "residual_std": 0.15,
        "residual_mean": 0.002
    }
}

趋势分析

trend_info = decomposer.analyze_trend()

# Returns:
{
    "direction": "increasing",   # "increasing", "decreasing", "flat"
    "slope": 0.023,
    "r_squared": 0.91,
    "change_points": [           # Detected trend changes
        {"index": 24, "date": "2023-01-01", "direction": "up"},
        {"index": 48, "date": "2025-01-01", "direction": "down"}
    ],
    "growth_rate": 0.028,        # Compound growth rate
    "volatility": 0.12
}

季节性分析

seasonal_info = decomposer.analyze_seasonality()

# Returns:
{
    "detected_period": 12,
    "strength": 0.72,
    "pattern": {
        1: {"value": 0.12, "label": "Jan", "rank": 3},
        2: {"value": -0.05, "label": "Feb", "rank": 8},
        ...
    },
    "peak_period": 12,           # Period with highest seasonal effect
    "trough_period": 2,          # Period with lowest seasonal effect
    "seasonal_range": 0.35       # Max - Min seasonal effect
}

周期检测

自动检测季节周期：

# Automatic detection using ACF
period = decomposer.detect_period()
print(f"Detected period: {period}")

# Or with decomposition
result = decomposer.decompose()  # Auto-detects period

异常检测

查找残差中的离群值：

anomalies = decomposer.detect_anomalies(threshold=2.0)

# Returns DataFrame with anomalous points:
#    date        value    residual    zscore    anomaly_type
# 0  2023-03-15  1250.5   450.2       3.2       high
# 1  2023-08-22  320.1    -380.5      -2.8      low

基础预测

# Trend extrapolation
forecast = decomposer.forecast(periods=12, method="trend")

# Seasonal naive (last season's values)
forecast = decomposer.forecast(periods=12, method="seasonal_naive")

# Trend + Seasonal
forecast = decomposer.forecast(periods=12, method="combined")

# Returns:
#    date        forecast    lower_bound    upper_bound
# 0  2024-01-01  1050.2      920.5          1180.0
# 1  2024-02-01  1080.5      945.2          1215.8

可视化

成分图
decomposer.plot_components("components.png")

生成一个包含 4 个面板的图：

  1. 原始序列
  2. 趋势
  3. 季节性
  4. 残差
ACF/PACF 图
decomposer.plot_acf_pacf("acf_pacf.png", lags=40)

自相关函数和偏自相关函数。

季节性图
decomposer.plot_seasonal("seasonal.png")

按周期显示季节性效应的柱状图。

示例工作流

销售分析
decomposer = TimeSeriesDecomposer()
decomposer.load_csv("monthly_sales.csv", "month", "revenue")

# Auto-detect and decompose
result = decomposer.decompose()

# Understand patterns
print(f"Trend: {decomposer.analyze_trend()['direction']}")
print(f"Peak season: Month {decomposer.analyze_seasonality()['peak_period']}")

# Plot
decomposer.plot_components("sales_analysis.png")
异常检测
decomposer = TimeSeriesDecomposer()
decomposer.load_csv("daily_metrics.csv", "date", "pageviews")
decomposer.decompose(period=7)  # Weekly pattern

# Find unusual days
anomalies = decomposer.detect_anomalies(threshold=2.5)
print(f"Found {len(anomalies)} anomalous days")
考虑季节性的预测
decomposer = TimeSeriesDecomposer()
decomposer.load_csv("quarterly_data.csv", "quarter", "value")
decomposer.decompose(period=4, model="multiplicative")

# Forecast next year
forecast = decomposer.forecast(periods=4, method="combined")
print(forecast)

依赖项

  • pandas>=2.0.0
  • numpy>=1.24.0
  • scipy>=1.10.0
  • statsmodels>=0.14.0
  • matplotlib>=3.7.0

评论

支持 Markdown