獲取內(nèi)容

股票名稱、縮寫闻鉴、領(lǐng)域及K線（日 周 月 季 年）

import os
import struct
import pandas as pd


# 負責讀取本地通達信數(shù)據(jù)
class TDXLoader:
    def __init__(self):
        # TDX_DIR 為通達信根目錄
        self.TDX_DIR = os.environ.get("Tdx")
        self.__base_all = self._get_base_all()

    def resolve(self, *args, **kwargs):
        return os.path.join(self.TDX_DIR, *args, **kwargs)

    # 獲取行業(yè)數(shù)據(jù)
    def _get_tdxhy(self):
        tdxhy_path = self.resolve(r"T0002\hq_cache\tdxhy.cfg")
        names = "market symbol industry_code idontcare1 idontcare2 industry_second_code".split()
        usecols = "market symbol industry_code industry_second_code".split()
        tdxhy = pd.read_csv(
            tdxhy_path,
            sep="|",
            names=names,
            usecols=usecols,
            dtype={"symbol": str},
        )
        return tdxhy

    # 行業(yè)代碼碼值
    def _get_incon(self):
        incon_path = self.resolve(r"incon.dat")
        incon = pd.read_csv(incon_path, encoding="gb2312", names=["index"])
        fx0 = lambda x: x.split("|")[0] if "|" in x else ""
        incon["industry_code"] = incon["index"].apply(fx0)
        fx1 = lambda x: x.split("|")[1] if "|" in x else ""
        incon["industry_name"] = incon["index"].apply(fx1)
        usecols = ["industry_code", "industry_name"]
        return incon[usecols]

    # 股票代碼對應(yīng)拼音縮寫
    def _read_tnf(self, path):
        market = path.split(".")[0][-3:]
        with open(path, "rb") as f:
            buff = f.read()

        data = buff[50:]
        l = len(data) // 314
        fx = lambda x: str(x, encoding="gbk").strip("\x00")
        sm = {"szm": ("00", "30"), "shm": ("60", "68")}

        stocks = []
        for x in [data[i * 314 : (i + 1) * 314] for i in range(l)]:
            code = fx(x[:6])
            if code.startswith(sm[market]):
                name = fx(x[23:41])
                shortcode = fx(x[285:293])

                stocks += [[code, name, shortcode]]
        return stocks

    # 股票K線數(shù)據(jù)源文件
    def _get_szshm(self):
        szm_path = self.resolve(r"T0002\hq_cache\szm.tnf")
        shm_path = self.resolve(r"T0002\hq_cache\shm.tnf")

        szm = self._read_tnf(szm_path)
        shm = self._read_tnf(shm_path)

        stocks = pd.DataFrame(szm + shm, columns=["symbol", "name", "shortcode"])
        return stocks

    # 整合基本數(shù)據(jù)
    def _get_base_all(self):
        tdxhy = self._get_tdxhy()
        incon = self._get_incon()
        szshm = self._get_szshm()

        base = pd.merge(szshm, tdxhy, how="left", on="symbol")
        base = pd.merge(base, incon, how="left", on="industry_code")
        base = pd.merge(
            base,
            incon,
            how="left",
            left_on="industry_second_code",
            right_on="industry_code",
        )

        fx = lambda x: ".sh" if x else ".sz"
        base["ts_code"] = base["symbol"] + base["market"].apply(fx)

        base.rename(
            columns={
                "industry_name_x": "industry_name",
                "industry_name_y": "industry_detail",
            },
            inplace=True,
        )
        usecols = "ts_code symbol name shortcode industry_name industry_detail".split()
        return base[usecols]

    # 讀取K線源文件
    def _read_kline(self, filepath):
        with open(filepath, "rb") as f:
            usecols = "trade_date open high low close amount vol openinterest".split()
            buffers = []
            while True:
                buffer = f.read(32)
                if not buffer:
                    break
                buffer = struct.unpack("lllllfll", buffer)
                buffers.append(buffer)
            kline = pd.DataFrame(buffers, columns=usecols)

        kline["trade_date"] = kline["trade_date"].astype(str)

        price_columns = ["open", "high", "low", "close"]
        kline[price_columns] = kline[price_columns].apply(lambda x: x / 100)
        return kline

    # 獲取基本數(shù)據(jù)
    def get_base_all(self):
        return self.__base_all

    # 獲取日K線數(shù)據(jù)
    def get_kline_daily(self, ts_code):
        filename = ts_code.split(".")[1] + ts_code.split(".")[0] + ".day"
        filepath = self.resolve("vipdoc", ts_code.split(".")[1], "lday", filename)
        kline = self._read_kline(filepath)
        kline["ts_code"] = ts_code
        kline.index = pd.to_datetime(kline["trade_date"])
        kline.index.name = "index"
        kline = kline.rename(columns={"vol": "volume"})
        usecols = (
            "ts_code trade_date open high low close amount volume openinterest".split()
        )
        return kline[usecols]


if __name__ == "__main__":
    loader = TDXLoader()
    # 獲取所有股票基本數(shù)據(jù)
    base_all = loader.get_base_all()
    # 獲取單股日K線數(shù)據(jù)
    kline = loader.get_kline_daily("600645.sh")
    print(kline)

走緩存，并加工出K線的各個周期

單單運行腳本緩存是無效的睁枕，需要放在Web服務(wù)或桌面應(yīng)用中

from store.loader import TDXLoader as Loader
from cacheout import LFUCache

from loguru import logger as log


# 負責緩存
class Cache(object):
    def __init__(self):
        self.__loader = Loader()
        self.cache = LFUCache()

        self.rule = {
            "ts_code": "last",
            "trade_date": "last",
            "open": "first",
            "high": "max",
            "low": "min",
            "close": "last",
            "volume": "sum",
            "openinterest": "last",
        }

    # 從緩存獲取日K線盗胀，沒有緩存就從Loader讀取
    def _get_kline_daily(self, ts_code):
        if self.cache.has(ts_code):
            kline = self.cache.get(ts_code)
        else:
            kline = self.__loader.get_kline_daily(ts_code)
            self.cache.set(ts_code, kline)
            log.info(f"{ts_code} K線數(shù)據(jù)已緩存")
        return kline

    # 按 日 周 月 季 年 獲取K線數(shù)據(jù)艘蹋，對應(yīng)周期為 D W M Q A。day week month quarter annual
    def _get_kline_by_period(self, ts_code, period="D"):
        kline = self._get_kline_daily(ts_code)

        kline = kline.resample(period).agg(self.rule).dropna()
        kline = kline.sort_values(by="trade_date", ascending=False)
        kline = kline.reset_index(drop=True)

        return kline

    # 按天數(shù)分割獲取K線數(shù)據(jù)票灰，days等于30則代表30天為一個周期
    def _get_kline_by_days(self, ts_code, days=1):
        kline = self._get_kline_daily(ts_code)

        kline = kline.sort_values(by="trade_date", ascending=False)
        kline = kline.reset_index(drop=True).sort_values(by="trade_date")
        kline = kline.groupby(lambda x: x // days).agg(self.rule)
        kline = kline.sort_values(by="trade_date", ascending=False)

        return kline

    # 獲取基本數(shù)據(jù)的函數(shù)女阀，傳股票代碼進來就獲取一只股票的，不傳參就獲取所有股票的
    def get_base(self, ts_code=None):
        data = self.__loader.get_base_all()
        if ts_code:
            base = data[data["ts_code"] == ts_code]
        else:
            base = data
        return base

    # 整合 _get_kline_by_period 和 _get_kline_by_days
    def get_kline(self, ts_code, param):
        if isinstance(param, int):
            return self._get_kline_by_days(ts_code, days=param)
        else:
            return self._get_kline_by_period(ts_code, period=param)


datas = Cache()


if __name__ == "__main__":
    # 獲取所有股票基本數(shù)據(jù)
    base_all = datas.get_base()
    # 獲取單股基本數(shù)據(jù)
    base_all = datas.get_base("600571.sh")
    # 獲取單股日線數(shù)據(jù)
    kline = datas.get_kline("600571.sh", "D")
    # 獲取單股周線數(shù)據(jù)
    kline = datas.get_kline("600571.sh", "W")
    # 獲取單股月線數(shù)據(jù)
    kline = datas.get_kline("600571.sh", "M")
    # 獲取單股季線數(shù)據(jù)
    kline = datas.get_kline("600571.sh", "Q")
    # 獲取單股年線數(shù)據(jù)
    kline = datas.get_kline("600571.sh", "A")
    # 獲取單股以5天為周期的數(shù)據(jù)
    kline = datas.get_kline("600571.sh", 5)
    # 獲取單股以30天為周期的數(shù)據(jù)
    kline = datas.get_kline("600571.sh", 30)