🎄
Frosty Friday #85 ASOF JOIN

2025/12/25に公開
 Frosty Fridayに挑戦Frosty FridayはSnowflakeのコミュニティチャレンジです。

100本ノック的にSnowflakeの基礎から応用までを勉強することができます。
今回はその中からWeek85「ASOF JOIN」に挑戦します。

この課題は金融市場をテーマに、時系列の取引データと見積もりデータを結合するというチャレンジです。

https://frostyfriday.org/blog/2024/03/15/week-85-intermediate/

 準備まずはセットアップコードを実行します。
create or replace temporary table trade (
    type VARCHAR,
    id INT,
    ticker VARCHAR,
    datetime TIMESTAMP_TZ,
    price FLOAT,
    volume INT
);

create or replace temporary table quotes(
    type VARCHAR,
    id INT,
    ticker VARCHAR,
    datetime TIMESTAMP_TZ,
    ask FLOAT,
    bid FLOAT
);
insert into trade values
    ('trade', 2, 'AAPL', '2020-01-06 09:00:30.000+09:00', 305, 1),  
    ('trade', 2, 'AAPL', '2020-01-06 09:01:00.000+09:00', 310, 2),  
    ('trade', 2, 'AAPL', '2020-01-06 09:01:30.000+09:00', 308, 1),  
    ('trade', 3, 'GOOGL', '2020-01-06 09:02:00.000+09:00', 1500, 2),  
    ('trade', 3, 'GOOGL', '2020-01-06 09:03:00.000+09:00', 1520, 3),  
    ('trade', 3, 'GOOGL', '2020-01-06 09:03:30.000+09:00', 1515, 1); 

insert into quotes values
    ('quote', 2, 'AAPL', '2020-01-06 08:59:59.999+09:00', 305, 304),  
    ('quote', 2, 'AAPL', '2020-01-06 09:02:00.000+09:00', 311, 309),  
    ('quote', 3, 'GOOGL', '2020-01-06 09:01:00.000+09:00', 1490, 1485),  
    ('quote', 3, 'GOOGL', '2020-01-06 09:04:00.000+09:00', 1530, 1528);  

insert into quotes values
    ('quote', 2, 'AAPL', '2020-01-06 09:00:30.000+09:00', 307, 305),  
    ('quote', 2, 'AAPL', '2020-01-06 09:01:30.000+09:00', 308, 306),  
    ('quote', 3, 'GOOGL', '2020-01-06 09:02:00.000+09:00', 1502, 1498),  
    ('quote', 3, 'GOOGL', '2020-01-06 09:03:30.000+09:00', 1518, 1513);  
二つのテーブルができあがりました。

金融取引を表すtradeテーブル



見積もりを表すquoteテーブル


 解答二つのテーブルをJoinしたいのですが、キーはIDとDATETIMEの複合キーです。

IDは常に一意に定まりますが、DATETIMEはトランザクションのタイミングなので、テーブル毎に値が変わっていて通常のJoinでは結合できません。

お題は「各取引記録(trade)と取引時点より以前に利用可能な最新の見積もり(quotes)を結合する」です。

こんなときにASOF JOINが利用できます。

基本構文はこんな感じです。
select ...
from table_A
asof join table_B
match_condition <時刻の結合条件>
on <時刻以外の列の結合条件>
通常のJoinにプラスしてmatch_conditionを設定することができます。
それでは解答です。
select
    T.type,
    T.id,
    T.ticker,
    T.datetime,
    T.price,
    T.volume,
    Q.type,
    Q.datetime,
    Q.ask,
    Q.bid
from trade as T
asof join quotes as Q
match_condition (T.datetime >= Q.datetime) -- Tのdatetimeに最も近いQのdatetimeをjoin
on T.id = Q.id
order by T.id, T.datetime;
match_conditionとして(T.datetime >= Q.datetime)を指定します。

これにより左側のテーブルのdatetimeより古く、最も近いdatetimeを持つレコードをJoinできます。

match_conditionには不等号を使用できますが、==は指定できません。これでは普通のJoinですからね…​…​
結果はこのようになります



tradeに対してquotesを紐づけることができました。

 ASOF JOINが使われるところASOF JOINの仕組みがわかっていても、なんとなく使いにくいなと感じるかもしれません。

実はSnowflakeの機能でASOF JOINが使われているところがあります。

それは特徴量ストアです。

特徴量ストアは機械学習で使用する特徴量を保存・管理することができる機能です。

保存するデータ(ビュー)は予めJoinに使用するキーを指定しておく必要があり、時系列データの場合はASOF JOINが実行されています。
※特徴量ストアについてはこちらで解説しています。
特徴量ストアからデータを取得するときは、欲しいデータのキーをspineデータフレームとして保持しておき、generate_training_set()を使って特徴量を含んだデータフレームを作成します。

この時に実行されるASOF JOINの結合方向と条件を確認してみます。

Snowflake MLライブラリはOSSとしてリポジトリが公開されています。
https://github.com/snowflakedb/snowflake-ml-python/blob/release_1.9.2/snowflake/ml/feature_store/feature_store.py#L1943-L1955

1953行目にmatch_conditionがありますね。

左側にはspineデータフレーム、右側には特徴量ビューが来ます。

したがってgenerate_training_set()で得られるデータフレームにはspineデータフレームで指定した時系列キーより古いデータのうち最も時間が近いデータが結合されます。
あまり需要は無いかもしれませんが、最後に今回の問題を特徴量ストアを使って解いてみます。
from snowflake.snowpark.context import get_active_session
session = get_active_session()

# 特徴量ストアを初期化
from snowflake.ml.feature_store import FeatureStore, CreationMode

FS_DB = "FROSTYFRIDAY"    
FS_NAME = "FS_WEEK85"        
FS_WH = session.get_current_warehouse()

## 初回は CREATE_IF_NOT_EXIST を指定するとスキーマが作成されます
fs = FeatureStore(
    session=session,
    database=FS_DB,
    name=FS_NAME,
    default_warehouse=FS_WH,
    creation_mode=CreationMode.CREATE_IF_NOT_EXIST
)

# エンティティを作成
from snowflake.ml.feature_store import Entity

id_entity = Entity(
    name="ID",
    join_keys=["ID"],
    desc="sample ID"
)
ts_entity = Entity(
    name="DATETIME",
    join_keys=["DATETIME"],
    desc="sample datetime"
)

## エンティティを Feature Store に登録
fs.register_entity(id_entity)
fs.register_entity(ts_entity)

# 特徴量ビューを登録
from snowflake.ml.feature_store import FeatureView

trade_df = session.table('frostyfriday.week85.trade')

trade_fv = FeatureView(
    name="FV_TRADE",
    entities=[id_entity, ts_entity],
    feature_df=trade_df
)
fs.register_feature_view(trade_fv, version='1.0')

quote_df = session.table('frostyfriday.week85.quotes')

quote_fv = FeatureView(
    name="FV_QUOTE",
    entities=[id_entity, ts_entity],
    feature_df=quote_df
)
fs.register_feature_view(quote_fv, version='1.0')

# データセットを作成（内部でASOF JOINが実行される）
spine_df = (
    session.table('frostyfriday.week85.trade')
    .select('id', 'datetime')
)
trade_fv = fs.get_feature_view('FV_TRADE', version='1.0')
quote_fv = fs.get_feature_view('FV_QUOTE', version='1.0')
training_df = fs.generate_training_set(
    spine_df=spine_df,                         
    features=[trade_fv, quote_fv],             
    spine_timestamp_col="DATETIME",        
    include_feature_view_timestamp_col=False
)
training_df.show(10)
無事データを取得できました。


 まとめ時系列データの分析でASOF JOINを使えば、絞り込み条件にbetweenを使わなくていいシチュエーションがあるかも！？
Frosty Fridayは毎回勉強になるお題ばかりですね！

皆さんもぜひ挑戦してみてください
Frosty Fridayに挑戦

準備

解答

ASOF JOINが使われるところ

まとめ

Discussion
