PythonでFXチャートを表示してみる Csv2WebAPI編

PythonでFXチャートを表示するのをGoogle Colabでやってみます。
https://colab.research.google.com

PythonでFXチャートを表示してみる API編
https://qiita.com/namikitakeo/items/c05cb9af05ac53c18dbf

Twelve DataのWebAPI実行回数には制限があります。過去データについては、アキシオリーのヒストリカルデータをWebAPI化して使おうと思います。

wget https://www.axiory.com/jp/assets/download/historical/mt4_standard/2022/USDJPY.zip

Twelve DataのWebAPIのURLに自作WebAPIを設定します。

import datetime
import pandas as pd
import mplfinance as mpf
from twelvedata import TDClient
from twelvedata.http_client import DefaultHttpClient

TWELVE_DATA_API_KEY = '****************************' # Twelve Data のAPIキー

#API_URL = 'https://api.twelvedata.com'
API_URL = 'https://***.mydns.jp' # Csv2WebAPIのURL

# ヒストリカルデータを取得
def get_historical_data(symbol, interval, outputsize, start_date, end_date, timezone):
#    td = TDClient(apikey = TWELVE_DATA_API_KEY)
    td = TDClient(apikey = TWELVE_DATA_API_KEY, http_client=DefaultHttpClient(API_URL))

    res = td.time_series(
              symbol = symbol,
              interval = interval,
              outputsize = outputsize,
              start_date = start_date,
              end_date = end_date,
              timezone = timezone
          ).as_json()

    df = pd.DataFrame(res).iloc[::-1].set_index('datetime').astype(float)
    df = df[df.index >= start_date]
    df.index = pd.to_datetime(df.index)

    return df

Apache2のProxy設定で、自作WebAPIをインターネットに公開します。
https://qiita.com/namikitakeo/items/02a1af116781622dff31

Csv2WebAPI.py

import json
from flask import *

app = Flask(__name__)

@app.route("/time_series")
def response_json():
   req = request.args
   symbol = req.get("symbol")
   interval = req.get("interval")
   outputsize = req.get("outputsize")
   start_date = req.get("start_date")
   s = r'{"meta":{"symbol":"'+symbol+'","interval":"'+interval+'","type":"Physical Currency"},"values":['
   filename = symbol[0:3]+symbol[4:7]+'_'+start_date[0:4]+'_'+start_date[5:7]+'.csv'
   start=start_date[0:10]
   output = 0
   max = int(outputsize)
   if interval=="1h":
       max *= 60
   if interval=="1day":
       max *= 60 * 24
   first=""
   last=""
   high=""
   low=""
   with open(filename, 'r') as f:
      for row in f:
         datetime=row.replace('.','-')
         hour=datetime[11:13]
         min=datetime[14:16]
         if datetime[0:10] < start:
            continue
         if output<max:
            s1=row.split(',')
            if interval=="1day":
               if (datetime[0:10]!=last):
                  last=datetime[0:10]
                  if (first == "1"):
                     s+='"high":"'+high+'",'
                     s+='"low":"'+low+'",'
                     s+='"close":"'+(s2[5])+'"},'
                     first = ""
                  s+='{"datetime":"'+datetime[0:10]+' '+datetime[11:16]+':00",'
                  s+='"open":"'+(s1[2])+'",'
                  high=(s1[3])
                  low=(s1[4])
                  first = "1"
            elif interval=="1h":
               if (hour!=last): 
                  last=hour
                  if (first == "1"):
                     s+='"high":"'+high+'",'
                     s+='"low":"'+low+'",'
                     s+='"close":"'+(s2[5])+'"},'
                     first = ""
                  s+='{"datetime":"'+datetime[0:10]+' '+datetime[11:16]+':00",'
                  s+='"open":"'+(s1[2])+'",'
                  high=(s1[3])
                  low=(s1[4])
                  first = "1"
            else:
               s+='{"datetime":"'+datetime[0:10]+' '+datetime[11:16]+':00",'
               s+='"open":"'+(s1[2])+'",'
               s+='"high":"'+(s1[3])+'",'
               s+='"low":"'+(s1[4])+'",'
               s+='"close":"'+(s1[5])+'"},'
            s2=s1
            if high<(s1[3]):
               high=(s1[3])
            if low>(s1[4]):
               low=(s1[4])
         output += 1
   if (interval=="1day" or interval=="1h"):
      s+='"high":"'+high+'",'
      s+='"low":"'+low+'",'
      s+='"close":"'+(s2[5])+'"},'
   s+='],"status":"ok"}'
   s3=s.replace(',]',']')
   data = json.loads(s3)
   data['values'] = sorted(data['values'], key=lambda k: k['datetime'], reverse=True)
   return (data)

if __name__ == '__main__':
   app.run()