bitflyer apiを作って、過去の約定データから、1週間分の1分足データを生成して、CSV保存してみた。

admin

5か月前

過去1ヶ月までみたい。
{“status”:-156,”error_message”:”Execution history is limited to the most recent 31 days.”,”data”:null}

import requests
import pandas as pd
from datetime import datetime, timedelta
import pytz
import time

class BitflyerAPI:
    def __init__(self, base_url='https://api.bitflyer.com'):
        self.base_url = base_url
    
    def executions(self, product_code, count=500, before=None):
        params = {
            'product_code': product_code,
            'count': count
        }
        if before:
            params['before'] = before
        response = requests.get(f'{self.base_url}/v1/getexecutions', params=params)
        return response.json() if response.status_code == 200 else []

api = BitflyerAPI()

# 1週間前の日時を取得（UTC）
utc = pytz.UTC
end_date = utc.localize(datetime.utcnow())
start_date = end_date - timedelta(days=7)

executions = []
count = 0
while True:
    if executions:
        last_id = executions[-1]['id']
        new_executions = api.executions(product_code="FX_BTC_JPY", count=500, before=last_id)
    else:
        new_executions = api.executions(product_code="FX_BTC_JPY", count=500)
    
    if not new_executions:
        break
    
    executions.extend(new_executions)
    count += len(new_executions)
    
    # 取得したデータの最初のtimestampを確認
    latest_timestamp = pd.to_datetime(executions[-1]['exec_date'], errors='coerce', utc=True)
    
    # 進捗状況を表示
    print(f"取得したデータ件数: {len(executions)}, 最新のデータ日時: {latest_timestamp}")
    
    # 1週間前までのデータが取得できたら終了
    if latest_timestamp <= start_date:
        break

    # リクエスト間の待機時間を設定（API制限に配慮）
    time.sleep(1)

if not executions:
    raise ValueError("APIからデータを取得できませんでした")

# DataFrameに変換
df = pd.DataFrame(executions)
df['timestamp'] = pd.to_datetime(df['exec_date'], errors='coerce', utc=True)
df.dropna(subset=['timestamp'], inplace=True)  # 'timestamp'がNaTの行を削除
df.set_index('timestamp', inplace=True)

# 1週間前のデータだけをフィルタリング
df = df[df.index >= start_date]

# 1分足データを生成
ohlcv = df['price'].resample('1min').ohlc()
ohlcv['volume'] = df['size'].resample('1min').sum()
ohlcv.dropna(inplace=True)

# CSVに保存
ohlcv_file = 'ohlcv_data.csv'
ohlcv.to_csv(ohlcv_file)

print(f"データの保存が完了しました: {ohlcv_file}")

import requests

import pandas as pd

from datetime import datetime, timedelta

import pytz

import time

class BitflyerAPI:

def __init__(self, base_url='https://api.bitflyer.com'):

self.base_url = base_url

def executions(self, product_code, count=500, before=None):

params = {

'product_code': product_code,

'count': count

}

if before:

params['before'] = before

response = requests.get(f'{self.base_url}/v1/getexecutions', params=params)

return response.json() if response.status_code == 200 else []

api = BitflyerAPI()

# 1週間前の日時を取得（UTC）

utc = pytz.UTC

end_date = utc.localize(datetime.utcnow())

start_date = end_date - timedelta(days=7)

executions = []

count = 0

while True:

if executions:

last_id = executions[-1]['id']

new_executions = api.executions(product_code="FX_BTC_JPY", count=500, before=last_id)

else:

new_executions = api.executions(product_code="FX_BTC_JPY", count=500)

if not new_executions:

break

executions.extend(new_executions)

count += len(new_executions)

# 取得したデータの最初のtimestampを確認

latest_timestamp = pd.to_datetime(executions[-1]['exec_date'], errors='coerce', utc=True)

# 進捗状況を表示

print(f"取得したデータ件数: {len(executions)}, 最新のデータ日時: {latest_timestamp}")

# 1週間前までのデータが取得できたら終了

if latest_timestamp <= start_date:

break

# リクエスト間の待機時間を設定（API制限に配慮）

time.sleep(1)

if not executions:

raise ValueError("APIからデータを取得できませんでした")

# DataFrameに変換

df = pd.DataFrame(executions)

df['timestamp'] = pd.to_datetime(df['exec_date'], errors='coerce', utc=True)

df.dropna(subset=['timestamp'], inplace=True) # 'timestamp'がNaTの行を削除

df.set_index('timestamp', inplace=True)

# 1週間前のデータだけをフィルタリング

df = df[df.index >= start_date]

# 1分足データを生成

ohlcv = df['price'].resample('1min').ohlc()

ohlcv['volume'] = df['size'].resample('1min').sum()

ohlcv.dropna(inplace=True)

# CSVに保存

ohlcv_file = 'ohlcv_data.csv'

ohlcv.to_csv(ohlcv_file)

print(f"データの保存が完了しました: {ohlcv_file}")

関連記事