一、接口快速上手1.1 代码格式韩国股票代码统一格式6 位数字 .KSKOSPI 和 KOSDAQ 共用这一格式。005930.KS 三星电子KOSPI 000660.KS SK 海力士KOSPI 035420.KS NAVERKOSPI 247540.KS 에코프로비엠KOSDAQ注意虽然数字段有一定规律大致上 KOSPI 蓝筹集中在0xxxxxKOSDAQ 成长股在2xxxxx和3xxxxx但这不是绝对规则不要用来做板块判断。板块信息应通过独立的标的列表接口获取。1.2 三个核心 REST 接口实时成交明细importrequests HEADERS{apiKey:YOUR_API_KEY}#获取免费API KEY: www.infoway.iodefget_trade(codes:str)-list:urlfhttps://data.infoway.io/korea/batch_trade/{codes}returnrequests.get(url,headersHEADERS).json()[data]tradesget_trade(005930.KS,000660.KS)fortintrades:print(f{t[s]}: 价格{t[p]}, 成交量{t[v]}, 时间戳{t[t]}ms)字段含义s标的代码t成交时间毫秒时间戳p最新成交价韩元v成交量股vw成交额韩元td方向0未知1买入2卖出最优盘口defget_depth(codes:str)-list:urlfhttps://data.infoway.io/korea/batch_depth/{codes}returnrequests.get(url,headersHEADERS).json()[data]fordinget_depth(005930.KS):ask_p,ask_vd[a][0][0],d[a][1][0]bid_p,bid_vd[b][0][0],d[b][1][0]print(f{d[s]}: 卖一{ask_p}×{ask_v}买一{bid_p}×{bid_v})盘口返回的买卖各一档。a是卖盘b是买盘每个字段是两个子数组[[价格列表], [数量列表]]用下标对应。K 线defget_kline(codes:str,kline_type:int8,num:int10)-list:urlhttps://data.infoway.io/korea/v2/batch_klinepayload{klineType:kline_type,klineNum:num,codes:codes}returnrequests.post(url,jsonpayload,headers{**HEADERS,Content-Type:application/json}).json()[data]klineType取值11分钟、25分钟、315分钟、430分钟、51小时、62小时、74小时、8日K、9周K、10月K、11季K、12年K。二、重点注意事项2.1 时区直接写死 UTC9韩国标准时间KST固定为UTC9无夏令时。这一点比美股友好很多不需要每年处理两次时钟切换。但无夏令时正因为反直觉反而容易出错。如果你的代码里有类似根据月份判断时区偏移的逻辑在韩国数据上会产生错误。正确做法fromzoneinfoimportZoneInfofromdatetimeimportdatetime KSTZoneInfo(Asia/Seoul)defts_ms_to_kst(ts_ms:int)-datetime:将毫秒时间戳转换为 KST 时间。returndatetime.fromtimestamp(ts_ms/1000,tzKST)# 示例dtts_ms_to_kst(1735000000000)print(dt.strftime(%Y-%m-%d %H:%M:%S %Z))# 2024-12-24 08:26:40 KST用Asia/Seoul作为时区名让系统库处理历史时区记录不要手动加减 9 小时。另外注意行情接口返回的时间戳单位是毫秒t字段历史 K 线接口的翻页参数timestamp是秒。两处单位不同混用会导致翻页到错误位置。2.2 批量查询的两个隐藏限制批量查询时有两个约束文档里写了但容易在实际使用中忽略限制一单次请求最多 100 个标的REST 接口成交明细、盘口、K 线单次请求的codes参数最多携带 100 个标的。如果你要覆盖 KOSPI 全市场 940 只股票需要自己实现分批逻辑fromitertoolsimportislicedefchunked(iterable,n):ititer(iterable)whilebatch:list(islice(it,n)):yieldbatch all_codes[005930.KS,000660.KS,...]# 全部标的列表all_trades[]forbatchinchunked(all_codes,100):codes_str,.join(batch)all_trades.extend(get_trade(codes_str))限制二多标的 K 线klineNum最多填 2这个限制是很多人第一次批量拉 K 线时发现的当codes包含多个标的时klineNum最大值是2只返回最近 2 根 K 线。要获取某个标的较长时间段的 K 线必须单独查询# 正确单标的最多 500 根defget_kline_history(code:str,kline_type:int,num:int500):returnget_kline(code,kline_type,min(num,500))# 需要翻页时传入最早一根 K 线的秒时间戳减 1defget_kline_page(code:str,kline_type:int,before_ts_sec:int):urlhttps://data.infoway.io/korea/v2/batch_klinepayload{klineType:kline_type,klineNum:500,codes:code,timestamp:before_ts_sec-1}resprequests.post(url,jsonpayload,headers{**HEADERS,Content-Type:application/json})returnresp.json()[data][0][respList]2.3 ±30% 涨跌停不能套 A 股逻辑这是最容易出错的一点。韩国股票的单日涨跌幅限制是±30%远高于 A 股主板的 ±10%。如果你的代码有这样的判断# 错误A股逻辑套在韩股上会产生大量误报ifabs(change_pct)0.095:flag_as_limit_hit()韩国正确的判断方式KOREA_LIMIT_THRESHOLD0.295# 留 0.5% 的浮动空间defis_limit_hit(current_price:float,prev_close:float)-str:change(current_price-prev_close)/prev_closeifchangeKOREA_LIMIT_THRESHOLD:return涨停ifchange-KOREA_LIMIT_THRESHOLD:return跌停returnK 线返回的pc字段涨跌幅百分比形式也可以直接用于判断forbarinkline_data:pctfloat(bar[pc])# 已是百分比数字如 -12.5 表示跌 12.5%ifabs(pct)29.5:print(f{bar[s]}触及涨跌停涨跌幅:{pct}%)值得一提的是韩国 KOSDAQ 生物医药和二线科技股在重大消息新药临床数据、监管审批结果、并购公告发布后直接从开盘打到涨跌停并不罕见。如果你的系统有基于价格变动幅度的告警在接入韩国市场时一定要重新校准阈值否则会产生大量噪声。2.4 集合竞价时段的数据处理韩国交易所的集合竞价分两段时段KST北京时间开盘集合竞价08:30–09:0007:30–08:00收盘集合竞价15:20–15:3014:20–14:30集合竞价期间行情接口依然会推送数据但成交价是竞价撮合中产生的预计开盘/收盘价可能和正式交易开始后的首笔成交有差距。如果你的策略逻辑依赖开盘价建议等 09:00KST正式交易开始后的第一笔成交数据而不是直接取 08:30 之后最早的推送价格。一个简单的过滤函数fromdatetimeimporttime MARKET_OPEN_KSTtime(9,0)MARKET_CLOSE_KSTtime(15,20)defis_continuous_trading(ts_ms:int)-bool:判断时间戳是否在正式连续交易时段09:00–15:20 KST。dt_kstts_ms_to_kst(ts_ms)tdt_kst.time()returnMARKET_OPEN_KSTtMARKET_CLOSE_KST2.5 非交易时段的 WebSocket 行为韩国市场收盘后KST 18:00 以后WebSocket 连接保持正常心跳会正常收到响应但不再有行情推送。这是正常现象不是连接异常。在策略代码里要区分连接正常但无数据和连接断开两种状态避免把盘后静默误判为数据管道故障importtimeclassDataHealthChecker:def__init__(self,timeout_sec:int120):self.last_data_tstime.time()self.timeouttimeout_secdefon_message(self):self.last_data_tstime.time()defis_stale(self)-bool:数据静默超过 timeout 秒。returntime.time()-self.last_data_tsself.timeoutdefis_trading_hours(self)-bool:判断当前是否在交易时段KST 09:00–15:30含集合竞价。fromdatetimeimportdatetime now_kstdatetime.now(tzKST)tnow_kst.time()weekdaynow_kst.weekday()ifweekday5:# 周六日returnFalsereturntime(8,30)ttime(15,30)defshould_alert(self)-bool:returnself.is_stale()andself.is_trading_hours()只在交易时段数据静默时才触发告警非交易时段的静默忽略。2.6 WebSocket 断线重连后必须重新订阅这一点在文档里有说明但实际操作中还是经常忘记WebSocket 断线重连后之前的订阅状态不会自动恢复服务端是无状态的。重连后如果没有重新发送订阅请求连接建立成功但不会有任何行情推送。正确做法是把建立连接和发送订阅绑定在一起asyncdef_connect_once(self):asyncwithwebsockets.connect(self.ws_url)asws:self.wswsawaitself._subscribe_all()# 每次连接后立即重新订阅self._start_heartbeat()asyncformessageinws:self._on_message(message)不要把订阅请求放在只执行一次的初始化函数里每次重连都要重新调用。2.7 请求频率控制REST 接口有频率限制具体额度取决于套餐。如果同时并发请求多个批次容易触发限频返回 429。一个简单的令牌桶实现importthreadingimporttimeas_timeclassRateLimiter:def__init__(self,calls_per_second:float):self.interval1.0/calls_per_second self._lockthreading.Lock()self._last_call0.0defwait(self):withself._lock:now_time.monotonic()elapsednow-self._last_callifelapsedself.interval:_time.sleep(self.interval-elapsed)self._last_call_time.monotonic()# 使用示例每秒最多 5 次请求limiterRateLimiter(calls_per_second5)forbatchinchunked(all_codes,100):limiter.wait()tradesget_trade(,.join(batch))三、小科普韩国为什么不用夏令时顺带提一个容易被忽略的历史背景了解这个对时区处理会有更清晰的认知。韩国曾经实行过夏令时。1987 年首尔奥运会前夕韩国试验性地引入夏令时目的是与国际时间接轨、减少夏季用电高峰。但实施后社会反响很差上班族发现夏天要在天还没亮的时候上班农村地区的作息根本不配合时钟变化加上实际节电效果存疑1988 年奥运会结束后当年就正式废除。此后韩国再未重新实行夏令时KST 固定在 UTC9。相比之下相邻的日本同为 UTC9在二战前后也短暂实行过夏令时同样以废除告终。东亚主要市场中国、日本、韩国如今全部没有夏令时和美股比起来时区处理要简单很多。四、完整示例带异常处理的生产级查询脚本把上面的注意事项整合成一个实用的多标的行情采集脚本importrequestsimporttimefromdatetimeimportdatetime,timeasdtimefromzoneinfoimportZoneInfofromitertoolsimportislice KSTZoneInfo(Asia/Seoul)HEADERS{apiKey:YOUR_API_KEY}KOREA_LIMIT_PCT29.5defchunked(iterable,n):ititer(iterable)whilebatch:list(islice(it,n)):yieldbatchdefis_trading_hours()-bool:nowdatetime.now(tzKST)ifnow.weekday()5:returnFalsetnow.time()returndtime(9,0)tdtime(15,30)defsafe_get(url:str,params:dictNone,retries:int3)-dict|None:forattemptinrange(retries):try:resprequests.get(url,headersHEADERS,paramsparams,timeout10)resp.raise_for_status()returnresp.json()exceptrequests.HTTPErrorase:ifresp.status_code429:wait2**attemptprint(f限频{wait}s 后重试...)time.sleep(wait)else:print(fHTTP 错误{resp.status_code}:{e})returnNoneexceptrequests.RequestExceptionase:print(f请求失败{attempt1}/{retries}:{e})time.sleep(1)returnNonedeffetch_all_trades(codes:list[str])-dict[str,dict]:result{}forbatchinchunked(codes,100):codes_str,.join(batch)datasafe_get(fhttps://data.infoway.io/korea/batch_trade/{codes_str})ifdataanddataindata:foritemindata[data]:result[item[s]]item time.sleep(0.2)# 批次间间隔避免触发限频returnresultdefcheck_limit_status(trade:dict,prev_close:float)-str:currentfloat(trade[p])pct(current-prev_close)/prev_close*100ifpctKOREA_LIMIT_PCT:returnf涨停 ({pct:.1f}%)ifpct-KOREA_LIMIT_PCT:returnf跌停 ({pct:.1f}%)returnf{pct:.2f}%defmain():watchlist[005930.KS,# 三星电子000660.KS,# SK 海力士035420.KS,# NAVER207940.KS,# 三星生物373220.KS,# LG 新能源]ifnotis_trading_hours():print(f当前不在交易时段KST{datetime.now(tzKST).strftime(%H:%M)}退出。)returnprint(f开始采集共{len(watchlist)}个标的当前 KST:{datetime.now(tzKST).strftime(%H:%M:%S)})tradesfetch_all_trades(watchlist)forcode,tradeintrades.items():ts_dtdatetime.fromtimestamp(int(trade[t])/1000,tzKST)direction{0:—,1:↑买,2:↓卖}.get(trade.get(td,0),—)print(f{code:12s}| 价格: ₩{float(trade[p]):10,.0f}f| 量:{float(trade[v]):8,.0f}f| 方向:{direction}f| 时间:{ts_dt.strftime(%H:%M:%S)})if__name____main__:main()五、常见问题盘口接口的a和b数组里有时会出现空列表怎么处理**非交易时段或标的未有成交时买卖盘可能为空。解析前先检查长度askslist(zip(d[a][0],d[a][1]))ifd[a]andlen(d[a][0])0else[]bidslist(zip(d[b][0],d[b][1]))ifd[b]andlen(d[b][0])0else[]K 线的pc字段是基于什么计算的pcpercent change是相对于上一根 K 线收盘价的涨跌幅单位是百分比如-5.2表示跌 5.2%。对于日K这就是昨日收盘价到当日当前价的涨跌幅。注意当日 K 线在收盘前是未完成状态pc会随最新成交价实时变化。WebSocket 心跳间隔设多少合适建议30 秒发一次心跳协议号 10010。间隔太长超过 60 秒可能被服务端主动断开间隔太短会增加不必要的网络开销。能否通过 K 线接口判断某一天是否是交易日**可以间接判断查询日K如果某个日期没有对应的 K 线记录则说明当天是非交易日节假日或周末。但更推荐的做法是维护一份韩国交易日历KRX 官网每年发布次年的交易日历可以提前下载缓存。免费套餐可以订阅多少个韩国标的的 WebSocket免费套餐全市场包括韩国、港股、日股等所有市场共 10 个 WebSocket 订阅额度。如果只需要少量核心标的做测试免费套餐完全够用。批量查询时哪些标的代码查不到数据代码停牌或退市时REST 接口可能不返回该标的的记录不是报错而是data数组里缺少该条目。处理批量结果时要用返回结果里的s字段实际返回的代码作为索引而不是假设顺序和请求顺序一致。小结接入韩国行情 API 本身不难但有几个细节值得在上线前专门确认时区写死 UTC9、批量 K 线的klineNum限制、±30% 涨跌停的判断逻辑、非交易时段的静默处理以及断线重连后的重新订阅。这些地方踩一次坑改起来都不复杂但如果在生产环境才发现排查起来会比较费时。做多市场数据接入最省力的方式是为每个市场单独维护一份差异备忘而不是假设所有市场的规则都和你最熟悉的那个市场一样。韩国市场整体对开发者比较友好时区简单、无午休、代码格式统一值得放进多市场数据源的候选清单里。