関数ベースの時系列

関数ベースの時系列を使用すると、関数で定義した Python ロジックを使って数値時系列を生成および変換できます。Foundry は、時系列同期を定義することなく、関数の出力を時系列として扱います。

機能

• カスタム分析： 数値時系列を生成する Python 関数を作成します。statsmodels や Prophet などの任意のライブラリ、または独自のコードを使用して、高度な分析を実行できます。
• Foundry での直接統合： 関数の出力を Quiver で直接活用し、互換性と組み合わせの柔軟性を確保しながら、リサンプリング、数式、結合、時系列検索などの操作を適用できます。
• オンデマンドのデータアクセス： データを事前に実体化することなく、外部 API やサービスからデータを簡単に取り込み、迅速なプロトタイピングと動的な分析を実現できます。
• パラメーター化されたシナリオ： 制御設定や予測期間などのカスタム入力を渡し、生成された複数の時系列を左右配置で比較できます。
• スケーラブルな本番ワークフロー： 組み込みの結果キャッシュとストリーミング実行を活用し、インタラクションの応答性を維持しながら大規模な出力を処理できます。

以下の例では、Workshop モジュールで関数ベースの時系列を使用し、週次予測をリアルタイムで計算しています。オペレーターは、機械の制御設定の変更が性能予測にどのような影響を与えるかについて、さまざまなシナリオをシミュレーションできます。

仕組み

関数ベースの時系列には、シリアル化された数値時系列を返す Python の Foundry 関数が必要です。データをクエリすると、Quiver は指定されたパラメーターで関数を呼び出し、クエリ時に時系列を動的に生成します。この出力は Foundry 内で通常の時系列と同様に扱われるため、さらに操作や可視化を適用できます。

この統合により、カスタムモデルやその場で実行する分析をダッシュボードに簡単に組み込めるようになり、柔軟でモジュール化された時系列ワークフローを実現できます。

ユースケースの例

• 予測とシミュレーション： Prophet や statsmodels など、好みのライブラリを使用して予測を生成したり、Foundry で構築したモデルを使用して将来の傾向を分析し、シナリオプランニングを実行したりできます。
• 多変量分析： 複数のセンサーやメトリクスの出力を組み合わせ、包括的な時系列分析を実現できます。
• モデルの迅速な反復改善： 中間データセットやデータパイプラインを管理することなく、カスタムモデルを迅速に反復改善できます。

例：Prophet を使用して予測する

以下の例では、関数ベースの時系列と Prophet ↗ を組み合わせて予測する方法を示します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from functions.api import function
from timeseries_sdk.types import TimeSeries
from ontology_sdk.ontology.objects import Machine
from prophet import Prophet

@function
def performance_prophet_forecast(
    machine: Machine,
    periods: int = 96,      # 将来のステップ数
    freq: str = "15min",    # サンプリング頻度
) -> list[bytes]:
    """
    Prophet を使用してマシンの performance_score を予測します。
    """
    df = machine.performance_score.to_pandas(all_time=True)
    if df.empty:
        return TimeSeries.serialize(df)

    df = df.rename(columns={"timestamp": "ds", "value": "y"}).sort_values("ds")
    model = Prophet().fit(df)
    future = model.make_future_dataframe(periods=periods, freq=freq, include_history=True)
    forecast = model.predict(future)
    out = forecast[["ds", "yhat"]].rename(columns={"ds": "timestamp", "yhat": "value"})
    return TimeSeries.serialize(out)