使用pytdx获取历史股票行情

先看效果

获取从2010年01月01日-2024年09月30日的股票数据

pytdx基础

https://blog.csdn.net/firexiaHouse/article/details/142687052?spm=1001.2014.3001.5501

获取历史股票行情

    def get_k_data(self, code, start, end):
        logging.info("获取历史股票行情")
        if self.api.connect(self.ip, int(self.port)):
            df = self.api.get_k_data(code, start, end)
            return df.to_dict('records') or []
        return []

将历史数据存入数据库

    def create_history_individual_stock(self, stock_id, date_from, date_to, page):
        """

        """
        start = timeit.default_timer()
        if not date_from:
            date_from = date.today() - timedelta(days=30)
        else:
            date_from = datetime.strptime(date_from, '%Y-%m-%d').date()
        if not date_to:
            date_to = date.today()
        else:
            date_to = datetime.strptime(date_to, '%Y-%m-%d').date()
        if date_from > date_to:
            return
        tdx_client = self.env['tdx.service'].get_tdx_client()
        history_k_data = tdx_client.get_k_data(
            stock_id.code, date_from.strftime('%Y-%m-%d'), date_to.strftime('%Y-%m-%d')
        )
        individual_stock_data = []
        for security_data in history_k_data:
            if security_data is None:
                continue
            stock_date = datetime.strptime(security_data.get('date'), '%Y-%m-%d').date()
            stock_vals = self.prepare_history_individual_stock_vals(security_data)
            stock_vals.update({
                'stock_id': stock_id.id,
                'stock_date': stock_date,
            })
            individual_stock_data.append(stock_vals)
        self.env['individual.stock'].create(individual_stock_data)
        end = timeit.default_timer()
        logging.info('获取历史股票行情, 第%s页，总计条数:%s, 总计用时：%s', page, len(individual_stock_data),
                     end - start)