FoundryTS で時系列データを操作する

FoundryTS は、時系列データに対してクエリを実行する Python ライブラリで、 Code Repositories、 Code Workbooks、Code Workspaces と統合されています。

詳細は 時系列のドキュメンテーション をご覧ください。

Code Repositories クイックスタートガイド

まず始めに、すでに Python コードリポジトリのセットアップ を完了していることを確認してください。

ライブラリ パネルを使用して、foundryts、transforms-timeseries、transforms-objects のライブラリをリポジトリに追加します。

次に、以下の手順に従って、クエリ対象のオブジェクトタイプをリポジトリにインポートし、プロジェクト参照を更新します：

1. リポジトリの 設定 タブに移動します。
2. オントロジーを選択します。
3. オブジェクトタイプを追加します。

株価データの取り扱い

この例では、Ticker name プロパティを持つ Stock series オブジェクトタイプから始めます。目的は、Technology セクターのすべてのシリーズを見つけて、その時間範囲を計算することです。

まず、変換の入力と出力を定義します。Stock series オブジェクトタイプをオブジェクト入力として宣言し、時間シリーズ同期を時間シリーズ入力として宣言します。

@transform(
    output=Output("/Users/jdoe/foundryts-test-energy-sector"),  # 出力先のディレクトリを指定
    ts=TimeSeriesInput('ri.time-series-catalog.main.sync.6bdbda27-29...'),  # 時系列データの入力を指定
    objects=ObjectInput(
        object_type_rid='ri.ontology.main.object-type.4168ed49-00...',  # オブジェクトタイプのリソースIDを指定
        ontology_rid='ri.ontology.main.ontology.00000000-00...',  # オントロジーのリソースIDを指定
        ontology_branch_rid='ri.ontology.main.branch.00000000-00...'  # オントロジーのブランチのリソースIDを指定
    )
)

ここで、変換関数を定義し、FoundryTS のインスタンスを初期化します。この関数は、オブジェクトタイプ、時間系列同期、および出力を引数として取ることに注意してください。

def compute(ctx, ts, objects, output):
    # FoundryTS()関数を用いて新たな時間系列オブジェクトを作成します。
    fts = FoundryTS()

次に、Technology セクターの Stock series オブジェクトを検索します。検索結果の各シリーズを、その時間範囲（最も早い時点と最も遅い時点のタイムスタンプ）にマップします。

    # エネルギーセクターに関連する時系列データを検索するコード
    search_result = fts.search.series(
        (ontology('sector') == 'energy'),  # エネルギーセクターに関連するデータを検索条件に設定
        object_types=['timeseries-demo-stock-series']  # 時系列データのタイプを指定
    ).map(F.time_extent())  # 検索結果の各時系列データの時間範囲を取得

最後に、データフレームを出力データセットに書き込みます。

    df = search_result.to_dataframe()  # search_resultをデータフレームに変換します
    output.write_dataframe(df)  # データフレームを出力します

すべてを組み合わせると、完成した変換は次のようになります。

from transforms.api import transform, Output
from transforms.timeseries import TimeSeriesInput
from foundryts import FoundryTS
from foundryts.search import ontology
import foundryts.functions as F
from transforms.objects import ObjectInput


@transform(
    output=Output("/Users/jdoe/foundryts-test-energy-sector"), # 出力先のパスを指定します
    ts=TimeSeriesInput('ri.time-series-catalog.main.sync.6bdbda27-29...'), # 時系列データの入力を指定します
    objects=ObjectInput(
        object_type_rid='ri.ontology.main.object-type.4168ed49-00...', # オブジェクトのタイプを指定します
        ontology_rid='ri.ontology.main.ontology.00000000-00...', # オントロジーを指定します
        ontology_branch_rid='ri.ontology.main.branch.00000000-00...' # オントロジーのブランチを指定します
    )
)
def compute(ctx, ts, objects, output):
    fts = FoundryTS() # FoundryTSオブジェクトを作成します

    # 'sector'が'energy'である時系列データを検索し、その時間範囲をマッピングします
    search_result = fts.search.series(
        (ontology('sector') == 'energy'),
        object_types=['timeseries-demo-stock-series']
    ).map(F.time_extent())

    df = search_result.to_dataframe() # 検索結果をデータフレームに変換します
    output.write_dataframe(df) # データフレームを出力します

これが出力データセットの見た目です。