KIS Open Trading API：量化交易工具箱，从API对接到策略回测全流程解析
1. 项目概述：一个为AI与开发者设计的量化交易工具箱
如果你是一名对量化交易感兴趣的Python开发者，或者正尝试用ChatGPT、Claude这类大语言模型（LLM）来辅助构建交易策略，那么你很可能面临一个共同的起点难题：如何与券商的交易系统对接？面对官方API文档里繁杂的接口说明、晦涩的参数和缺乏实战示例的困境，从零搭建一个稳定可靠的交易接口层，其工作量往往不亚于策略开发本身。
今天要深入剖析的，正是为解决这一痛点而生的项目——韩国投资证券（Korea Investment & Securities, KIS）官方维护的 open-trading-api 开源仓库。这远不止是一套简单的API调用示例，而是一个从 环境配置、数据获取、策略构建、历史回测到实盘交易 的完整工具链。它的核心价值在于，通过高度模块化、文档清晰的Python代码，将券商API的复杂性封装起来，让开发者能专注于策略逻辑本身。无论是想快速验证一个想法的新手，还是需要构建稳健自动化交易系统的资深量化研究员，这个项目都提供了一个极佳的起点和参考框架。
项目最鲜明的特色是其“双轨制”设计：一方面，它为LLM智能体（如ChatGPT的Code Interpreter）提供了极度细粒度、功能单一化的代码样本（ examples_llm/ ），方便AI理解并生成特定功能的代码；另一方面，它为人类开发者准备了按金融产品（如国内股票、海外期货期权）分类的、功能聚合的实战示例（ examples_user/ ）。这种设计思维本身就体现了对现代开发范式（人机协作）的深刻理解。此外，项目还集成了基于Web的图形化策略构建器（ strategy_builder/ ）和基于QuantConnect Lean引擎的本地化回测系统（ backtester/ ），形成了一套开箱即用的轻量级量化投研环境。
注意 ：本文所有讨论均基于该开源项目的公开代码与文档，旨在进行技术方案解析与学习交流。所有交易API的使用必须遵守韩国投资证券的相关服务条款，并且在实际交易中，开发者需对自身代码逻辑的完备性与风险承担全部责任。
2. 核心架构与设计哲学解析
2.1 双轨制代码结构：服务于人与AI的两种思维模式
项目的目录结构清晰地反映了其设计目标。理解这种设计，是高效使用它的关键。
examples_llm/ 目录：面向AI的“原子化”指令集 这个目录下的代码组织方式，非常契合大语言模型处理任务的特点。每个API功能都被拆解到最细的粒度，一个文件夹只做一件事。例如，查询国内股票当前价格的完整路径是 examples_llm/domestic_stock/inquire_price/ 。在这个文件夹里，你通常只会找到两个文件：
inquire_price.py: 一个极简的函数，只负责调用“查询价格”这一个API，并返回原始数据。chk_inquire_price.py: 一个调用上述函数并打印或简单验证结果的脚本。
这种结构的优势在于“信息密度低，意图明确”。当你在ChatGPT中提问“如何使用KIS API查询三星电子的股价？”时，你可以直接将 inquire_price.py 的代码作为上下文提供给AI。AI能够毫无干扰地理解这个片段的功能，并在此基础上为你修改参数或组合逻辑。它避免了让AI去一个庞大的、混合了多种功能的文件中进行“阅读理解”和“代码抽取”，极大提高了指令遵循的准确率。
examples_user/ 目录：面向开发者的“模块化”工具箱 对于人类开发者而言，我们更习惯按领域或功能模块来查找和使用代码。因此，这个目录采用了不同的组织方式。以 domestic_stock （国内股票）为例，你会看到：
domestic_stock_functions.py: 一个集成了该品类下所有REST API函数（如查询价格、查询余额、下单、撤单等）的“超级模块”。domestic_stock_examples.py: 一个展示了如何使用上述集成模块中各个函数的示例脚本。- 对应的
_ws.py文件则是为WebSocket实时行情接口准备的。
这种结构的好处是“开箱即用，便于集成”。当你想开发一个股票自动交易程序时，你只需要 import domestic_stock_functions ，然后像调用本地库一样使用 inquire_price() , order_buy() , order_sell() 等函数。所有的认证、请求构造、错误处理等底层细节都被封装在 kis_auth.py 和这些函数内部。
设计背后的考量 ：这种双轨制设计并非多余，它精准地应对了两种不同的工作流。LLM辅助编程时，需要“喂给”它干净、单一的样本；而人类在构建系统时，则需要高内聚、低耦合的模块。项目通过维护两套代码，虽然增加了少许维护成本，但为两种主流用户群体都提供了最优体验。
2.2 认证与配置中心化： kis_auth.py 的核心角色
几乎所有金融API的使用，第一步也是最繁琐的一步就是认证。 kis_auth.py 文件是这个项目的“中枢神经”，它优雅地解决了这个问题。
核心机制解析 ：
- 配置管理 ：它约定了一个固定的配置文件路径（
~/KIS/config/kis_devlp.yaml），用于集中管理所有敏感信息（App Key, Secret, 账户号等）。这种做法避免了将密钥硬编码在代码中，符合安全最佳实践。通过环境变量或命令行参数可以覆盖此路径，提供了灵活性。 - 令牌生命周期管理 ：KIS的API使用OAuth 2.0类似的访问令牌机制。
kis_auth.py中的auth()函数不仅负责获取令牌，更重要的是，它内部实现了令牌的缓存和自动刷新逻辑。它会检查内存中已有的令牌是否即将过期（通常有一个安全缓冲期，如到期前5分钟），如果即将过期或不存在，则自动发起新的令牌申请请求。这意味着开发者在调用业务API时，几乎不需要关心令牌问题，就像使用一个永远有效的会话一样。 - 环境切换抽象 ：它通过
svr参数（如“prod”实盘，“vps”模拟盘）和product参数（如“01”综合账户）来抽象不同的交易环境。底层代码会根据这些参数自动选择正确的App Key、Secret和API网关地址。这使得同一套代码无需修改就能在模拟和实盘环境间无缝切换，对于策略测试至关重要。
实操心得：配置文件的处理 项目建议将自带的 kis_devlp.yaml 模板复制到 ~/KIS/config/ 目录下修改。这里有一个更工程化的建议：你可以将此配置目录的路径加入项目的 .gitignore 文件，确保你的密钥不会意外提交到代码仓库。对于团队协作，可以提交一个 kis_devlp.example.yaml 模板文件，而将真实的配置文件排除在版本控制之外。
2.3 策略构建与回测的闭环： strategy_builder 与 backtester
这是项目超越简单API封装的亮点，它提供了一个从想法到验证的微型工作流。
strategy_builder ：低代码策略设计器 这是一个基于Web的图形化界面。你不需要编写任何代码，可以通过拖拽和配置的方式，组合超过80个技术指标（如移动平均线、RSI、布林带等）来创建交易信号。项目内置了10个经典策略预设，例如“金叉死叉”、“动量突破”、“52周新高”等，对于初学者是极好的学习起点。
其核心输出是一个 .kis.yaml 文件。这个YAML文件用一种结构化的格式完整描述了一个交易策略的所有规则和参数。例如，一个简单的双均线金叉策略在YAML中可能被定义为：当5日均线上穿20日均线时，生成 BUY 信号；当下穿时，生成 SELL 信号。这种格式既便于人类阅读，也便于程序解析，成为了连接策略设计和回测执行的“协议”。
backtester ：本地化专业回测引擎 项目没有重复造轮子去实现一个回测系统，而是巧妙地集成了QuantConnect的Lean引擎——这是一个在专业量化领域被广泛使用、久经考验的开源回测框架。项目通过Docker将Lean引擎封装起来，并制作了适配层，使其能够读取上一步生成的 .kis.yaml 策略文件。
回测过程大致如下：
- 从
strategy_builder导出.kis.yaml文件。 - 将该文件放入
backtester的指定目录。 - 启动Docker容器，回测引擎会解析YAML文件，将其转换为Lean引擎能理解的算法。
- 引擎从数据源（项目可能内置或需要配置历史数据）加载指定时间段的历史行情数据。
- 按照策略规则在历史数据上模拟交易，并计算一系列绩效指标：如总收益率、年化收益率、夏普比率、最大回撤、胜率等。
- 生成可视化的HTML报告，直观展示资金曲线、买卖点、持仓变化等。
这个闭环的价值 ：它让个人开发者或小团队能够以极低的成本，拥有接近专业机构的策略研究流程。你可以在图形界面中快速迭代想法，然后立即进行严谨的历史回测，避免了在策略逻辑和回测代码之间来回切换的割裂感。
3. 从零开始的完整实操指南
3.1 环境准备与依赖安装
Python版本管理建议 项目要求Python 3.11+。强烈建议使用 pyenv （Mac/Linux）或 pyenv-win （Windows）来管理多个Python版本。这可以确保你的开发环境独立且纯净，不会影响系统自带的Python。
# 以Mac/Linux为例，使用pyenv安装指定版本
pyenv install 3.11.5
pyenv local 3.11.5 # 在当前目录下使用该版本
使用uv进行极速依赖管理 项目推荐使用 uv 这个用Rust编写的超快Python包管理工具。其速度远超传统的 pip ，并且能创建可复现的依赖环境。
# 安装uv（如果尚未安装）
# 在终端中执行以下命令之一：
# curl -LsSf https://astral.sh/uv/install.sh | sh # Mac/Linux
# powershell -c "irm https://astral.sh/uv/install.ps1 | iex" # Windows PowerShell
# 克隆项目并进入目录
git clone https://github.com/koreainvestment/open-trading-api.git
cd open-trading-api
# 一键同步所有依赖，uv会根据 pyproject.toml 创建虚拟环境并安装包
uv sync
执行 uv sync 后，所有必需的库（如 requests , websocket-client , pandas , numpy 等）都会自动安装到一个独立的虚拟环境中。后续运行脚本时，使用 uv run python script.py 即可在该环境中执行。
3.2 KIS API服务申请与密钥获取
这是接入真实交易功能的必要前提，过程需要一些耐心。
- 账户准备 ：你需要拥有一个韩国投资证券的实际交易账户。通常这需要联系相关的国际客户服务部门完成开户。
- 服务开通 ：登录韩国投资证券的HTS（Home Trading System）或手机App，找到“Open API服务”或“开发者服务”申请入口。提交申请，通常需要阅读并同意相关协议。
- 获取密钥 ：申请通过后，在开发者门户（KIS Developers Portal）中，你可以创建“应用”。每个应用都会生成一对密钥：
- App Key ：相当于你的应用用户名，公开或半公开。
- App Secret ：相当于你的应用密码，必须严格保密。 重要 ：你需要分别申请 模拟交易（Virtual Private Server, VPS） 和 实盘交易（Production） 两套密钥。模拟环境用于无风险测试，有调用频率限制；实盘环境用于真实交易。
3.3 配置文件 kis_devlp.yaml 的详细配置
按照项目要求创建配置文件：
mkdir -p ~/KIS/config
cp open-trading-api/kis_devlp.yaml ~/KIS/config/
然后用文本编辑器打开 ~/KIS/config/kis_devlp.yaml ，进行如下配置：
# ！！！请用你申请到的真实密钥替换下面的示例内容 ！！！
# 实盘交易环境密钥
my_app: "PSaaxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
my_sec: "RVEzenxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx=="
# 模拟交易环境密钥
paper_app: "PSaxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
paper_sec: "RVEzxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx=="
# 你的HTS登录ID（通常是你的账户ID或注册手机号）
my_htsid: "your_hts_login_id"
# 账户号码（前8位）
my_acct_stock: "12345678" # 股票账户前8位
my_acct_future: "87654321" # 期货期权账户前8位（如有）
my_paper_stock: "55556666" # 模拟交易股票账户前8位
my_paper_future: "66667777" # 模拟交易期货期权账户前8位（如有）
# 账户产品代码（后2位），最常用的是综合账户
my_prod: "01" # 01: 综合账户
# User-Agent，保持默认即可，用于模拟浏览器请求
my_agent: "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
关键细节 ：
my_acct_stock和my_paper_stock填的是账户号的 前8位 。完整的韩国证券账户通常是10位数字，例如12345678-01。这里12345678是前8位，-01是产品代码（my_prod）。务必确认你填写的账户号与当前登录HTS所见的账户号前8位一致。
3.4 运行你的第一个API调用：查询股价
让我们从最简单的功能开始，验证整个环境是否畅通。我们使用 examples_user 目录下的集成示例。
-
定位并查看示例文件 ：
cd open-trading-api/examples_user/domestic_stock用编辑器打开
domestic_stock_examples.py。你会看到一个包含了多个函数示例的长脚本。 -
简化脚本以进行测试 ：为了避免一次性运行所有示例，我们创建一个新的测试文件，或修改原文件。建议新建一个
test_first.py：# test_first.py import sys import logging sys.path.extend(['..', '.']) # 添加路径，以便能导入上级目录的kis_auth import kis_auth as ka from domestic_stock_functions import inquire_price # 只导入我们需要的函数 # 设置日志，方便查看过程 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) def main(): # 1. 认证（默认使用模拟环境vps和综合账户01） logger.info("正在初始化认证...") ka.auth(svr="vps", product="01") # 使用模拟环境 trenv = ka.getTREnv() # 获取当前交易环境信息 logger.info(f"认证成功，当前环境: {trenv}") # 2. 调用API查询三星电子(005930)的当前价格 logger.info("正在查询三星电子(005930)当前价格...") try: # 参数解释： # env_dv: 环境区分，'real'表示实时（非测试）环境，这里用模拟盘也是'real' # fid_cond_mrkt_div_code: 市场区分代码，'J'代表韩国综合股指市场（KOSPI） # fid_input_iscd: 股票代码，'005930'是三星电子 result = inquire_price( env_dv="real", fid_cond_mrkt_div_code="J", fid_input_iscd="005930" ) # 3. 处理结果 logger.info("查询成功！") # result 通常是一个字典，包含API返回的所有字段 # 我们提取几个关键信息 if result and 'output' in result: output = result['output'] stock_name = output.get('hts_kor_isnm', 'N/A') # 股票名称 current_price = output.get('stck_prpr', 'N/A') # 当前价格 price_change = output.get('prdy_vrss', 'N/A') # 较前日变动 change_rate = output.get('prdy_ctrt', 'N/A') # 变动率 print(f"\n=== 查询结果 ===") print(f"股票名称: {stock_name}") print(f"当前价格: {current_price} 韩元") print(f"涨跌额: {price_change} 韩元") print(f"涨跌幅: {change_rate} %") else: logger.error(f"API返回异常结构: {result}") except Exception as e: logger.error(f"API调用失败: {e}", exc_info=True) if __name__ == "__main__": main() -
运行测试 ：
uv run python test_first.py如果一切配置正确，你将看到类似以下的输出：
2023-10-27 14:30:00,123 - INFO - 正在初始化认证... 2023-10-27 14:30:00,456 - INFO - 认证成功，当前环境: {'svr': 'vps', 'product': '01'} 2023-10-27 14:30:00,789 - INFO - 正在查询三星电子(005930)当前价格... 2023-10-27 14:30:01,234 - INFO - 查询成功！ === 查询结果 === 股票名称: 삼성전자 当前价格: 80500 涨跌额: 1500 涨跌幅: 1.90恭喜！这证明你的Python环境、API密钥、账户配置全部正确，已经成功连接到了韩国投资证券的模拟交易系统。
4. 核心功能模块深度解析与实战
4.1 国内股票交易全流程示例
一个完整的自动化交易程序，通常包含“数据获取 -> 决策 -> 下单 -> 风控 -> 查询”的循环。我们以国内股票为例，拆解这个流程。
4.1.1 获取实时与历史数据 除了基础的当前价查询（ inquire_price ），API还提供深度行情、分时图、历史K线等。
- 实时报价（WebSocket） ：对于高频策略，轮询REST API效率太低。项目提供了WebSocket示例。关键步骤是：
- 调用
ka.auth_ws()获取WebSocket专用密钥。 - 创建
KISWebSocket对象并连接。 - 使用
subscribe方法订阅感兴趣的股票代码和行情类型（如实时成交、十档报价）。 - 在回调函数中处理推送过来的数据。
# 摘自 domestic_stock_examples_ws.py 的简化版 ka.auth() ka.auth_ws() kws = ka.KISWebSocket(api_url="/tryitout") # 订阅三星电子和SK海力士的实时报价 kws.subscribe(request=asking_price_krx, data=["005930", "000660"]) # 此时 kws 会在后台接收数据，你需要编写处理逻辑 - 调用
- 历史K线数据 ：回测和策略分析的基础。相关函数可能需要组合调用，例如先获取股票列表，再循环获取每只股票的历史数据。注意API通常有调用频率和日期范围限制。
4.1.2 账户与持仓查询 在交易前和交易后，都需要了解账户状态。
inquire_balance: 查询账户的综合余额，包括总资产、现金、股票市值、可用资金、信用额度等。inquire_psbl_order: 查询可购买数量。这非常重要，它会根据你的现金、信用比例、单笔交易限额等规则，计算出你当前最多能买多少股指定的股票。inquire_ccnl: 查询当日成交明细。inquire_nccs: 查询未成交委托（挂单）。
4.1.3 委托下单与撤单 这是交易的核心。项目封装了 order_buy （买入）和 order_sell （卖出）函数。下单时需要仔细设置参数：
pdno: 股票代码。ord_qty: 委托数量。ord_unpr: 委托价格。如果是市价单，则传0。ord_dvsn: 委托类型代码 。这是最容易出错的地方之一。例如：“00”: 限价委托“01”: 市价委托“02”: 条件指定价委托“03”: 最佳价委托
sll_buy_dvsn: 买卖区分。“02”为买入，“01”为卖出。
一个限价买入的示例：
order_result = order_buy(
env_dv="real",
sll_buy_dvsn="02", # 买入
ord_dvsn="00", # 限价单
pdno="005930", # 三星电子
ord_qty=10, # 10股
ord_unpr=80000 # 限价80000韩元
)
print(f"委托结果: {order_result}")
# 成功会返回委托号('odno')，用于后续查询或撤单。
4.1.4 实战组合：一个简单的定时定投脚本 假设我们想每个交易日收盘前5分钟，自动买入固定金额的三星电子。我们可以将上述函数组合起来。
# auto_invest.py
import kis_auth as ka
from domestic_stock_functions import inquire_balance, inquire_psbl_order, order_buy
import schedule
import time
from datetime import datetime
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
TARGET_STOCK = "005930"
DAILY_INVEST_AMOUNT = 100000 # 每日投入10万韩元
def job():
logger.info("=== 执行定时定投任务 ===")
ka.auth(svr="vps", product="01") # 模拟环境
# 1. 查询当前股价
price_info = inquire_price(env_dv="real", fid_cond_mrkt_div_code="J", fid_input_iscd=TARGET_STOCK)
current_price = int(price_info['output']['stck_prpr'])
logger.info(f"目标股票当前价格: {current_price}")
# 2. 计算可买数量 (向下取整)
quantity = DAILY_INVEST_AMOUNT // current_price
if quantity == 0:
logger.warning(f"当前价格{current_price}过高，{DAILY_INVEST_AMOUNT}韩元不足以购买1股。")
return
# 3. 查询实际可购买数量（考虑手续费、最低交易单位等限制）
psbl_info = inquire_psbl_order(env_dv="real", sll_buy_dvsn="02", pdno=TARGET_STOCK, ord_unpr=current_price)
max_psbl_qty = int(psbl_info['output']['ord_psbl_qty'])
final_qty = min(quantity, max_psbl_qty)
if final_qty == 0:
logger.warning("账户可购买数量为0，可能现金不足或达到限额。")
return
# 4. 执行市价买入委托
logger.info(f"尝试市价买入 {TARGET_STOCK} {final_qty}股")
try:
order_result = order_buy(
env_dv="real",
sll_buy_dvsn="02",
ord_dvsn="01", # 市价单
pdno=TARGET_STOCK,
ord_qty=final_qty,
ord_unpr=0 # 市价单价格为0
)
if order_result['rt_cd'] == '0': # '0'通常代表成功
odno = order_result['output']['odno']
logger.info(f"委托成功！委托号: {odno}")
else:
logger.error(f"委托失败: {order_result['msg1']}")
except Exception as e:
logger.error(f"下单过程发生异常: {e}")
# 设定每个交易日 14:55 执行 (韩国时间收盘前5分钟)
# 注意：这里使用简单调度，实际应用应考虑节假日
schedule.every().day.at("14:55").do(job)
logger.info("定时定投程序已启动，等待执行时间...")
while True:
schedule.run_pending()
time.sleep(60) # 每分钟检查一次
重要提醒 ：这是一个极度简化的示例。真实环境中必须加入异常处理、网络重试、日志记录、风险检查（如单日亏损限额）、以及节假日判断等严密的风控逻辑。切勿直接用于实盘。
4.2 策略构建器（Strategy Builder）实战入门
图形化界面降低了策略设计的门槛。启动后，通常可以通过 http://localhost:3000 访问。
- 界面概览 ：左侧是指标库，中间是画布，右侧是参数面板。你可以从左侧拖拽指标（如“SMA简单移动平均线”）到画布上。
- 创建一个金叉策略 ：
- 拖拽两个“SMA”指标到画布，分别设置短周期（如5）和长周期（如20）。
- 拖拽一个“交叉”条件节点到画布。
- 将短周期SMA的输出线连接到交叉节点的“系列A”输入，长周期SMA连接到“系列B”输入。
- 在交叉节点上选择“A上穿B”（Golden Cross）。
- 将这个交叉节点的输出，连接到一个“信号生成”节点，并设置信号为“BUY”。
- 同理，可以再创建一个“A下穿B”（Death Cross）的条件，连接到“SELL”信号。
- 参数优化 ：你可以点击SMA指标，在右侧面板将周期参数从固定值改为一个范围（如短周期从3到10，长周期从15到30）。策略构建器在回测时可以对这个范围进行网格搜索，寻找最优参数组合。
- 导出策略 ：设计完成后，点击导出，生成
.kis.yaml文件。这个文件描述了整个策略的逻辑图。
4.3 回测器（Backtester）运行与报告解读
- 放置策略文件 ：将上一步导出的
.kis.yaml文件复制到backtester/strategies/目录下。 - 配置回测参数 ：通常需要编辑一个配置文件（如
config.yaml或通过UI），指定回测的标的（如005930）、时间范围（如2023-01-01到2023-12-31）、初始资金等。 - 运行回测 ：在
backtester目录下执行启动脚本（如./start.sh）。Docker容器会启动，加载策略和历史数据，开始计算。 - 分析报告 ：回测结束后，会在
backtester/results/目录下生成一个HTML报告。重点看以下指标：- Total Return（总收益） ：策略的绝对收益。
- Sharpe Ratio（夏普比率） ：衡量风险调整后收益，大于1通常算不错，大于2很好。
- Max Drawdown（最大回撤） ：策略运行期间，账户净值从高点回落的最大幅度。这个值越小越好，它反映了策略可能面临的最大亏损风险。
- Win Rate（胜率） ：盈利交易次数占总交易次数的比例。
- Profit Factor（盈亏比） ：总盈利 / 总亏损。大于1表示盈利超过亏损。
- 图表 ：观察资金曲线是否平滑上升，回撤期是否过长。买卖点标记是否合理。
回测的局限性 ：必须清醒认识到，历史回测表现优异绝不等于未来能盈利。回测存在“前视偏差”、“过度拟合”、“未考虑滑点和手续费”等问题。 backtester 的结果是一个重要的参考，但决策前还需进行严格的样本外测试和模拟盘验证。
5. 高级主题、常见陷阱与排查指南
5.1 WebSocket实时行情的高效管理与重连机制
在生产环境中使用WebSocket，稳定性至关重要。项目提供的示例可能只是一个简单连接，你需要自己构建健壮的重连逻辑。
import threading
import time
from websocket import WebSocketConnectionClosedException
class RobustKISWebSocket:
def __init__(self):
self.ws = None
self.connected = False
self.reconnect_interval = 5 # 重连等待秒数
self._stop_event = threading.Event()
def start(self):
"""启动WebSocket连接（在独立线程中）"""
self.thread = threading.Thread(target=self._run, daemon=True)
self.thread.start()
def _run(self):
while not self._stop_event.is_set():
try:
ka.auth()
ka.auth_ws() # 确保WS密钥有效
trenv = ka.getTREnv()
self.ws = ka.KISWebSocket(api_url="/tryitout")
# 设置回调函数
self.ws.on_message = self._on_message
self.ws.on_error = self._on_error
self.ws.on_close = self._on_close
self.ws.on_open = self._on_open
self.ws.run_forever() # 这是一个阻塞调用，直到连接断开
except (WebSocketConnectionClosedException, ConnectionError) as e:
logger.error(f"WebSocket连接异常断开: {e}")
except Exception as e:
logger.error(f"WebSocket运行发生未知错误: {e}")
# 连接断开后，等待重连
if not self._stop_event.is_set():
logger.info(f"{self.reconnect_interval}秒后尝试重连...")
time.sleep(self.reconnect_interval)
def _on_open(self, ws):
self.connected = True
logger.info("WebSocket连接已建立，重新订阅主题...")
# 连接建立后，重新订阅之前需要的股票代码
# 你需要维护一个订阅列表
# ws.subscribe(request=asking_price_krx, data=self.subscription_list)
def _on_message(self, ws, message):
# 处理实时行情数据
try:
data = json.loads(message)
# 你的业务逻辑，例如更新内存中的报价，触发交易条件等
self.process_market_data(data)
except json.JSONDecodeError as e:
logger.error(f"解析消息失败: {e}, 原始消息: {message}")
def _on_error(self, ws, error):
logger.error(f"WebSocket错误: {error}")
self.connected = False
def _on_close(self, ws, close_status_code, close_msg):
logger.warning(f"WebSocket连接关闭，状态码: {close_status_code}, 消息: {close_msg}")
self.connected = False
def stop(self):
"""优雅停止"""
self._stop_event.set()
if self.ws:
self.ws.close()
5.2 错误代码（RT_CD）详解与处理策略
API调用返回的JSON中， rt_cd 字段是关键。 “0” 表示成功，非 “0” 表示失败，具体的错误信息在 msg1 或 msg_cd 字段中。
| 常见错误码 | 含义 | 可能原因与处理建议 |
|---|---|---|
EGW00101 |
无效的App Key或Secret | 1. kis_devlp.yaml 中的密钥填写错误。 2. 密钥未激活或已过期。去开发者门户检查。 3. 模拟盘/实盘密钥用错了环境（ svr 参数）。 |
EGW00201 |
每秒/每日请求次数超限 | 模拟环境限制很严格。避免在循环中无休眠地高频调用API。加入 time.sleep() 控制频率，或切换至实盘环境测试。 |
OPSP0001 |
订单错误（如价格超出涨跌幅限制） | 委托价格不符合交易所规则。检查委托价格是否在当日涨跌停板范围内。 |
MKDG10010 |
账户余额不足 | 下单金额超过了可用资金。下单前务必调用 inquire_balance 和 inquire_psbl_order 进行校验。 |
OPSP0003 |
无效的股票代码 | 检查 pdno 参数是否正确，以及该代码在当前市场是否有效交易。 |
通用错误处理模式 ：
def safe_api_call(api_func, *args, **kwargs):
"""一个包装函数，用于安全调用API并处理常见错误"""
max_retries = 3
for attempt in range(max_retries):
try:
resp = api_func(*args, **kwargs)
if resp.get('rt_cd') == '0':
return resp # 成功
else:
error_msg = resp.get('msg1', 'Unknown error')
error_code = resp.get('msg_cd', 'Unknown code')
logger.warning(f"API调用业务失败 (尝试 {attempt+1}/{max_retries}): [{error_code}] {error_msg}")
# 针对特定错误码进行处理
if error_code == 'EGW00201':
logger.error("请求频率超限，等待60秒后重试...")
time.sleep(60)
continue # 重试
elif error_code in ['EGW00101', 'OPSP0003']:
# 认证或参数错误，重试无意义
raise ValueError(f"配置或参数错误: [{error_code}] {error_msg}")
else:
# 其他错误，可能重试也无用，直接退出或抛异常
break
except requests.exceptions.ConnectionError as e:
logger.error(f"网络连接错误 (尝试 {attempt+1}/{max_retries}): {e}")
time.sleep(5) # 等待后重试
except Exception as e:
logger.error(f"调用API时发生未知异常: {e}")
break # 未知异常，不重试
# 所有重试都失败
raise Exception(f"API调用失败，已达最大重试次数 {max_retries}")
5.3 性能优化与注意事项
- 令牌缓存 ：
kis_auth.py已经做了令牌缓存，但如果你在多进程或多线程环境下运行，需要确保令牌状态是线程/进程安全的，或者考虑使用外部的令牌存储（如Redis）。 - 请求合并与批处理 ：对于需要查询多只股票信息的情况，查看API文档是否有批量查询接口。如果没有，不要使用
for循环依次查询，这极易触发频率限制。应该使用异步IO（asyncio/aiohttp）或线程池来并发请求，并在每个请求间添加合理的间隔（如0.1秒）。 - 数据本地缓存 ：像股票基本信息、节假日日历等不常变的数据，应该一次性查询后缓存到本地文件或数据库，避免重复调用API。
- 日志与监控 ：自动化交易系统必须有详尽的日志记录，记录每一笔委托、成交、账户变动以及所有API调用和异常。建议使用
logging模块，并配置按日期和级别滚动存储日志文件。同时，可以设置关键指标（如账户净值、持仓风险）的监控告警。 - 模拟盘与实盘的差异 ：除了频率限制，模拟盘的数据流、成交机制（尤其是流动性）与实盘存在差异。在模拟盘表现良好的策略，在实盘可能会因为滑点、订单无法全部成交等问题而大打折扣。务必充分理解这种差异。
5.4 对接LLM（如ChatGPT）的高级技巧
examples_llm/ 目录是专门为此设计的。当你想让LLM帮你写一个交易脚本时，最佳实践是：
-
提供清晰的上下文 ：不要只说“帮我用KIS API下单”。应该提供具体的函数样本作为参考。
- 低效提示 ：“写一个用韩国投资证券API买股票的函数。”
- 高效提示 ：“请参考以下
inquire_price和order_buy函数的调用方式（它们来自KIS Open API的examples_llm目录），帮我写一个函数：当股价低于50日均线时，市价买入100股三星电子(005930)。请包含必要的认证和错误处理。” 同时，附上inquire_price.py和order_buy.py的代码片段。
-
任务分解 ：对于复杂任务，引导LLM分步完成。例如：“第一步，请先写一个函数获取三星电子的当前价和50日均价。第二步，写一个比较逻辑。第三步，整合下单逻辑。”
-
利用LLM进行代码解释与学习 ：你可以将
examples_user/中复杂的集成函数丢给LLM，让它为你生成注释或分解逻辑，帮助你快速理解代码库。
这个项目通过其清晰的结构，极大地优化了人机协作的体验，让开发者能更高效地利用AI辅助编程能力，将想法快速转化为可执行的交易代码。
码道开发者社区，聚焦华为云码道 CodeArts 代码智能体，沉淀 Agent、Skill、鸿蒙开发实战内容，供开发者查阅资料、交流技术、分享工程实践
更多推荐
17
0- 0
已为社区贡献196条内容
所有评论(0)