使用pytdx获取股票行情数据

先展示效果

股票清单

总共统计深市与沪市总计5397只股票信息。

股票行情

以下是详细的实现步骤

下载通达信金融终端

此步骤的目的是要获取pytdx插件中需要使用的服务器地址及端口


在通讯设置中可以查看当前信号较好的服务器地址，最终在安装目录下找到服务器地址。

安装插件pytdx

pip install pytdx

我使用的python版本为3.10

python代码实现

# -*- coding: utf-8 -*-
import logging
from pytdx.hq import TdxHq_APIclass Tdx_Client():def __init__(self, ip='111.229.247.189', port='7709'):self.ip = ipself.port = portself.api = TdxHq_API(auto_retry=True, raise_exception=False)self.api.connect(self.ip, int(self.port))def get_security_list(self, mkt_id, start):logging.info("获取股票列表")if self.api.connect(self.ip, int(self.port)):data = self.api.get_security_list(mkt_id, start)return datareturn Nonedef get_security_quotes(self, all_stock, code=None):logging.info("获取股票行情")if self.api.connect(self.ip, int(self.port)):data = self.api.get_security_quotes(all_stock, code)return data or []return []if __name__ == '__main__':tdx = Tdx_Client()tdx.get_security_quotes([(1, "600028")])

上面的代码展示的是如何调试接口，最后我们需要使用odoo开源平台将数据存入数据库，这里只提供关键代码；

同步股票列表

    def create_task_job(self, market_id, security_start, page):""""""start = timeit.default_timer()tdx_client = self.env['tdx.service'].get_tdx_client()data_list = tdx_client.get_security_list(market_id, security_start)if not data_list:returnstock_vals = []for data_item in data_list:if self.get_stock_sector(data_item['code']) == '其它':continuestock_id = self.env['stock.info'].search([('code', '=', data_item['code']), ('active', 'in', [True, False])], limit=1)if stock_id:continuestock_vals.append({'code': data_item['code'],'name': data_item['name'],'market_id': '沪市' if market_id == 1 else '深市','active': True})self.env['stock.info'].create(stock_vals)end = timeit.default_timer()logging.info('创建股票清单数据, 第%s页，总计条数:%s, 总计用时：%s', page, len(data_list),end - start)

同步股票行情

    def create_individual_stock(self, stock_ids_chunk, page):""""""start = timeit.default_timer()tdx_client = self.env['tdx.service'].get_tdx_client()stock_ids = self.env['stock.info'].browse(stock_ids_chunk)all_stock = []for stock_id in stock_ids:market_id = 1 if stock_id.market_id == '沪市' else 0all_stock.append((market_id, stock_id.code))security_quotes = tdx_client.get_security_quotes(all_stock)date_today = date.today()individual_stock_data = []for security_data in security_quotes:if security_data is None:continuestock_id = self.env['stock.info'].search([('code', '=', security_data['code']), ('active', 'in', [True, False])], limit=1)if not stock_id:continuestock_vals = self.prepare_individual_stock_vals(security_data)stock_vals.update({'stock_id': stock_id.id,'stock_date': date_today,})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)