from pytdx.hq import TdxHq_APIfrom pytdx.exhq import TdxExHq_APIfrom pytdx.params import TDXParamsif __name__ == "__main__": print("0. 测试开始") print("测试目标：获取期权的五档和1M数据，以及对应标的的当前价格") print() print("1. 测试获取标准行情的股票数据") print() try: std_api = TdxHq_API() std_ip = '109.244.35.28' std_port = 7709 with std_api.connect(std_ip, std_port): print("1.1 成功连接标准行情服务器") print("IP:" + std_ip + ", Port:" + str(std_port)) print() print("1.2 获取标的指数数据-000300") stock_data = std_api.get_security_quotes([(TDXParams.MARKET_SH, '000300')]) print(stock_data) print() print("1.3 获取标的股票数据-510300") print("注意：返回的价格为实际价格*10，需要自己对返回结果进行特殊处理") stock_data = std_api.get_security_quotes([ (TDXParams.MARKET_SH, '510300')]) print(stock_data) print() print() print("2. 测试获取扩展行情的期权数据") print() ext_api = TdxExHq_API() ext_ip = '109.244.35.23' ext_port = 7722 with ext_api.connect(ext_ip, ext_port): print("2.1 成功连接扩展行情服务器") print("IP:" + ext_ip + ", Port:" + str(ext_port)) print() print("2.2 获取扩展行情服务器支持的市场。可以从中查找市场ID以供获取行情之用。") all_markets = ext_api.get_markets() first_six_markets = all_markets[:5] for market in first_six_markets: print(market) print() print("2.3 获取上交所股票期权五档行情") call_2606_4800 = ext_api.get_instrument_quote(8, "10011316") print(call_2606_4800) print() print("2.4 获取上交所股票期权当天1分钟数据") all_minutes = ext_api.get_minute_time_data(8, "10011316") last_six_minutes = all_minutes[-6:] for minute_data in last_six_minutes: print(minute_data) print() print("3. 测试结束") print("测试结果：成功获取期权五档和1M数据，同时也能成功获取标的的当前价格") print() except Exception as e: print() print(f"测试失败。异常信息: {e}") print()
