从雅虎金融获取行情历史数据

fintools.py

#!/usr/bin/env python3
from argparse import ArgumentParser
import datetime
import requests



class FinTools:
    INTERVAL_CANDIDATES = ['1m', '2m', '5m', '15m', '30m', '60m', '90m', '1h', '1d', '5d', '1wk', '1mo', '3mo']
    def format_code(self, code_):
        code = code_.lower()
        if code == 'sse':
            return '000001.SS'
        return code_

    def download(self, code, start, end, interval=None):
        # start: 开始日期时间戳（秒）
        # end: 结束日期时间戳（秒）
        url = 'https://query1.finance.yahoo.com/v7/finance/download/' + self.format_code(code)
        if interval is None:
            interval = '1d'
        assert(interval in self.INTERVAL_CANDIDATES), '时间间隔非法'
        params = {
            'period1': int(start), # 开始时间
            'period2': int(end), # 结束时间
            'interval': interval, # 间隔1天，candidates = [1m, 2m, 5m, 15m, 30m, 60m, 90m, 1h, 1d, 5d, 1wk, 1mo, 3mo]
            'events': 'history'
        }
        response = requests.get(url, params=params)
        csv_text = response.text
        return csv_text


def ds2d(v):
    fs = [
        '%Y%m%d',
        '%Y-%m-%d'
    ]
    for f in fs:
        try:
            d = datetime.datetime.strptime(v, f)
            return d
        except:
            pass
    return datetime.datetime.now()

def main():
    now = datetime.datetime.now()
    today = datetime.datetime(now.year, now.month, now.day)
    start_time = today - datetime.timedelta(days=365 * 2)
    cli_parser = ArgumentParser()
    cli_parser.add_argument('--code', required=True, help='https://finance.yahoo.com/ 上的代码')
    cli_parser.add_argument('--start', required=False, type=ds2d, default=start_time)
    cli_parser.add_argument('--end', required=False, type=ds2d, default=today)
    cli_parser.add_argument('--interval', required=False, default='1d', help='默认获取每日的数据')
    cli_configure = cli_parser.parse_args()
    ft = FinTools()
    t = ft.download(
        cli_configure.code,
        cli_configure.start.timestamp(),
        cli_configure.end.timestamp(),
        cli_configure.interval
    )
    print(t)


if __name__ == '__main__':
    main()