pull-minute-k/get_min_data.py

from Ashare import *
from trade_day import *

import os
import time
import csv

date = time.strftime("%Y-%m-%d", time.localtime())
date = "2025-04-30"
save_path = os.path.split(os.path.realpath(__file__))[0]
save_path = os.path.join(save_path, date)
# 是否已经获取了数据
file_count = 0
if os.path.exists(save_path):
    for root, dirs, files in os.walk(save_path):
        file_count += len(files)
if file_count > 5000:
    print("already pulled taday's data!")
    exit()
# 今天是否是交易日
ok, trade_day = is_trade_day(date)
if not ok or not trade_day:
    print("taday is not trade day, skip")
    exit()

# 获取数据
stock_codes = {}
stock_list_file = os.path.join(os.path.split(os.path.realpath(__file__))[0], 'table-r.txt')
with open(stock_list_file, 'r', encoding='utf-8') as file:
    csv_reader = csv.DictReader(file)
    for row in csv_reader:
        code = row["code"]
        if code.startswith("sh") or code.startswith("sz"):
            stock_codes[code] = row["name"]

if not os.path.exists(save_path):
    os.makedirs(save_path)

for c, n in stock_codes.items():
    print(f"getting stock:{c} name:{n}")
    file_name = str(c) + "-" + str(n) + ".txt"
    file_name = file_name.replace("*", "#")
    file_name = os.path.join(save_path, file_name)
    if os.path.exists(file_name):
        print(f"file:{file_name} already exist, skip")
        continue
    try:
        df = get_price(c, frequency='1m', count=240)
        ctx = df.to_string()
    except:
        print("get price error")
        ctx = ""
    with open(file_name, "w", encoding="utf-8") as outfile:
        outfile.write(ctx)
    # time.sleep(0.01)