OpenDART API Skill

This skill helps you build Python scripts, bots, and analysis pipelines that consume the OpenDART (Digital Analysis & Retrieval of Transcripts) API operated by Korea's Financial Supervisory Service (FSS).

OpenDART provides electronic disclosure data for Korean public companies (including KOSPI, KOSDAQ, and KONEX listed firms). The English API base URL is:

https://engopendart.fss.or.kr/engapi/

All endpoints require a 40-character API authentication key (crtfc_key).

---

1. Installation

Via skills.sh (Recommended)

Install this skill into your AI agent with one command:

npx skills add dwk601/opendart-api-skill

This makes the skill available to Claude Code, Cursor, Codex, Gemini CLI, GitHub Copilot, OpenCode, and 50+ other agents that support the skills ecosystem.

Discover more: https://skills.sh/dwk601/opendart-api-skill/opendart-api

Manual Install

git clone https://github.com/dwk601/opendart-api-skill.git
cd opendart-api-skill
pip install aiohttp pandas

Then import the bundled client from scripts/opendart_client.py.

---

2. Authentication

Getting an API Key

1. Register at https://engopendart.fss.or.kr
2. Navigate to Request an Authentication Key
3. The issued key is a 40-character string

Best Practice: Use Environment Variables

import os

API_KEY = os.environ.get("OPENDART_API_KEY")
if not API_KEY:
    raise ValueError("Set OPENDART_API_KEY environment variable")

---

3. Core 3-Step Workflow

Almost every OpenDART workflow follows this pattern:

Step 1: Download Corporation Codes

OpenDART uses an 8-digit corp_code (not stock ticker). Download the master mapping once and cache it.

from opendart_client import OpenDartClient
import asyncio

async def main():
    client = OpenDartClient(api_key="YOUR_KEY")
    df_corps = await client.get_corp_codes()
    # Cache to avoid re-downloading
    df_corps.to_parquet("corp_codes.parquet")

asyncio.run(main())

Step 2: Resolve corp_code

# Find Samsung Electronics by stock code or name
samsung = df_corps[df_corps["stock_code"] == "005930"]
corp_code = samsung.iloc[0]["corp_code"]  # e.g., "00126380"

Step 3: Call the Target Endpoint

fin = await client.get_financials_single(
    corp_code=corp_code,
    bsns_year="2023",
    reprt_code="11011"  # Annual report
)
print(fin.head())

---

4. Async-First Patterns

OpenDART allows ~20,000 requests/day. For financial analysis pipelines that touch hundreds of companies, asyncio + aiohttp is essential.

Basic Async Session

import asyncio
from opendart_client import OpenDartClient

async def fetch_many():
    client = OpenDartClient(api_key="YOUR_KEY", max_concurrent=10)
    async with client:
        results = await asyncio.gather(
            client.get_financials_single("00126380", "2023", "11011"),
            client.get_financials_single("00126380", "2022", "11011"),
            client.get_financials_single("00126380", "2021", "11011"),
        )
    return results

Semaphore-Controlled Concurrency

The bundled client uses asyncio.Semaphore(max_concurrent) to stay within rate limits. Default is 10 concurrent requests.

client = OpenDartClient(api_key="YOUR_KEY", max_concurrent=20)

Sequential with Backoff

For polling bots, sequential requests with asyncio.sleep() are simpler and safer:

async def poll_loop(client, interval=86400):
    while True:
        filings = await client.search_filings(bgn_de="20240101")
        await process(filings)
        await asyncio.sleep(interval)

---

5. Bot Patterns for Financial Analysis

Pattern A: Polling Bot (New Filings)

Checks for new disclosures since a date and stores results.

import asyncio
from datetime import datetime, timedelta
from opendart_client import OpenDartClient

async def polling_bot(api_key, check_interval=86400):
    client = OpenDartClient(api_key=api_key)
    async with client:
        since = (datetime.now() - timedelta(days=1)).strftime("%Y%m%d")
        while True:
            filings = await client.search_filings(
                bgn_de=since,
                pblntf_ty="A",      # Periodic disclosures
                page_count=100
            )
            if not filings.empty:
                filings.to_parquet(f"filings_{since}.parquet")
                since = filings["rcept_dt"].max()
            await asyncio.sleep(check_interval)

# asyncio.run(polling_bot("YOUR_KEY"))

Pattern B: Batch Financial Pipeline

Downloads financials for many companies/years with controlled concurrency.

import asyncio
import pandas as pd
from opendart_client import OpenDartClient

async def batch_pipeline(api_key, corp_codes, years, reprt_code="11011"):
    client = OpenDartClient(api_key=api_key, max_concurrent=10)
    all_data = []

    async with client:
        tasks = [
            client.get_financials_single(corp_code, year, reprt_code)
            for corp_code in corp_codes
            for year in years
        ]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    for df in results:
        if isinstance(df, Exception):
            print(f"Error: {df}")
            continue
        if not df.empty:
            all_data.append(df)

    return pd.concat(all_data, ignore_index=True) if all_data else pd.DataFrame()

Pattern C: Multi-Company Monitor

Watch specific companies for new filings.

async def monitor_companies(api_key, corp_codes, callback, interval=3600):
    client = OpenDartClient(api_key=api_key)
    seen = set()

    async with client:
        while True:
            tasks = [client.search_filings(corp_code=c) for c in corp_codes]
            results = await asyncio.gather(*tasks)

            for corp_code, filings in zip(corp_codes, results):
                if filings.empty:
                    continue
                for _, row in filings.iterrows():
                    key = row["rcept_no"]
                    if key not in seen:
                        seen.add(key)
                        await callback(corp_code, row)

            await asyncio.sleep(interval)

---

6. DataFrame Output Conventions

The bundled client normalizes API responses into tidy pandas DataFrames.

List Endpoints (e.g., search_filings)

The list array in the JSON becomes a DataFrame with one row per item. Metadata (page_no, total_count) is stored as DataFrame attributes:

filings = await client.search_filings(corp_code="00126380")
print(filings.attrs.get("total_count"))

Single-Row Endpoints (e.g., get_corp_info)

Returns a pd.Series for easy field access:

info = await client.get_corp_info("00126380")
print(info["corp_name"], info["ceo_nm"])

Financial Statements

get_financials_all returns a DataFrame where each row is an account item:

account_nm	fs_div	sj_div	thstrm_nm	thstrm_amount	...
Revenue	CFS	IS	Current Term	3022314	...

---

7. Pagination & Rate Limits

Limit	Value
Daily requests	~20,000
Results per page	Max 100
Companies per multi request	Max 100

Auto-Pagination

The client handles pagination automatically. If you call search_filings with defaults, the client loops through all pages and concatenates them into a single DataFrame.

To fetch a specific page manually:

page_1 = await client.search_filings(
    corp_code="00126380", page_no=1, page_count=100
)

Rate Limit Tuning

If you hit 020 (call limit exceeded), reduce max_concurrent or add request_delay:

client = OpenDartClient(
    api_key="YOUR_KEY",
    max_concurrent=5,
    request_delay=0.2   # seconds between requests
)

---

8. Error Handling

Retry Strategy

The client retries automatically on transient errors with exponential backoff:

• Retry: 020 (rate limit), 800 (maintenance), network timeouts
• No retry: 010 (invalid key), 013 (no data), 021 (too many companies)

Common Errors in Financial Analysis Bots

Error	Cause	Fix
013 No data viewed	Company has no filing for that year/report	Skip or try adjacent year
021 Exceeded company limit	Multi request has >100 corp_codes	Batch into chunks of 100
100 Field value invalid	reprt_code or bsns_year wrong	Check references/disclosure-types.md
020 Call limit exceeded	Too many requests	Reduce concurrency, add delay

Graceful Degradation

Always use return_exceptions=True in asyncio.gather for batch pipelines:

results = await asyncio.gather(*tasks, return_exceptions=True)
for r in results:
    if isinstance(r, Exception):
        log_error(r)
        continue
    process(r)

---

9. Quick Reference: Most-Used Endpoints

Task	Method	Key Params
Search disclosures	search_filings	bgn_de, end_de, corp_code, pblntf_ty
Company overview	get_corp_info	corp_code
Single company financials	get_financials_single	corp_code, bsns_year, reprt_code
Multi-company comparison	get_financials_multi	corp_code (comma-separated up to 100), bsns_year, reprt_code
Full financial statements	get_financials_all	corp_code, bsns_year, reprt_code, fs_div (CFS/OFS)
Major shareholders	get_major_shareholders	corp_code, bsns_year, reprt_code
Executives	get_executives	corp_code, bsns_year, reprt_code
Dividends	get_dividends	corp_code, bsns_year, reprt_code
Download document	download_document	rcept_no, save path

Report Type Codes (reprt_code)

Code	Report
11013	1st Quarter Report
11012	Semi-annual Report
11014	3rd Quarter Report
11011	Annual Report (Business Report)

Financial Statement Division (fs_div)

Code	Meaning
CFS	Consolidated Financial Statements (연결재무제표)
OFS	Separate Financial Statements (재무제표)

---

10. Bundled Resources

scripts/opendart_client.py

A full runnable async client. Import and use directly:

from opendart_client import OpenDartClient

references/endpoints.md

Complete catalog of all API endpoints with parameters and response structures. Read this when you need details on a specific endpoint.

references/error-codes.md

Full error code table with handling recommendations.

references/disclosure-types.md

Lookup tables for pblntf_ty, pblntf_detail_ty, corp_cls, reprt_code, etc.

---

11. Example: Complete Financial Analysis Script

import asyncio
import os
import pandas as pd
from opendart_client import OpenDartClient

async def analyze_kospi_financials():
    api_key = os.environ["OPENDART_API_KEY"]
    client = OpenDartClient(api_key=api_key, max_concurrent=10)

    async with client:
        # 1. Load or download corp codes
        try:
            df_corps = pd.read_parquet("corp_codes.parquet")
        except FileNotFoundError:
            df_corps = await client.get_corp_codes()
            df_corps.to_parquet("corp_codes.parquet")

        # 2. Filter to KOSPI companies (corp_cls == 'Y')
        kospi = df_corps[df_corps["corp_cls"] == "Y"].head(20)
        codes = kospi["corp_code"].tolist()

        # 3. Batch fetch 2023 annual financials
        tasks = [
            client.get_financials_single(c, "2023", "11011")
            for c in codes
        ]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        # 4. Combine into analysis DataFrame
        dfs = [r for r in results if isinstance(r, pd.DataFrame) and not r.empty]
        if dfs:
            combined = pd.concat(dfs, ignore_index=True)
            combined.to_parquet("kospi_2023_financials.parquet")
            print(f"Fetched {len(combined)} rows for {len(dfs)} companies")
        else:
            print("No data fetched")

if __name__ == "__main__":
    asyncio.run(analyze_kospi_financials())

---

12. Tips for Financial Analysis Bots

1. Cache corp codes: The ZIP file is large. Download once per week.
2. Chunk multi requests: get_financials_multi accepts up to 100 codes.
3. Handle missing data: Not all companies file every report type.
4. Use CFS for listed companies: Consolidated statements are standard for KOSPI/KOSDAQ.
5. Store raw responses: Save parquet files before transformation for reproducibility.
6. Monitor API usage: The client logs requests; track against the 20K/day limit.

---

When the user asks for OpenDART code, always default to the async client pattern, pandas DataFrames, and the async with context manager. Point them to the bundled scripts/opendart_client.py for the full implementation.