korea-investment-stock 0.9.0

Pure Python wrapper for Korea Investment Securities OpenAPI

🚀 Korea Investment Stock

Pure Python wrapper for Korea Investment Securities OpenAPI

✨ NEW in v0.9.0: Environment variable auto-detection like boto3/Twilio!

✨ v0.8.0: Built-in rate limiting to prevent API errors

🎯 Purpose

A simple, transparent wrapper around the Korea Investment Securities OpenAPI. This library handles API authentication and request formatting, giving you direct access to the API responses without abstraction layers.

Philosophy: Keep It Simple

• Pure wrapper: Direct API access without magic
• Minimal dependencies: Only requests and pandas
• No abstraction: You get exactly what the API returns
• Implement your way: Add rate limiting, caching, retries as you need them

📦 Package Structure

korea_investment_stock/
├── korea_investment_stock.py   # Core API wrapper
├── cache/                       # Memory caching (opt-in)
│   ├── cache_manager.py
│   └── cached_korea_investment.py
├── rate_limit/                  # Rate limiting (opt-in)
│   ├── rate_limiter.py
│   └── rate_limited_korea_investment.py
└── token_storage/               # Token persistence (opt-in)
    ├── token_storage.py         # FileTokenStorage
    └── token_storage.py         # RedisTokenStorage

Design Pattern: Wrapper-based architecture

• Core KoreaInvestment class unchanged
• Optional features as opt-in wrappers
• Stack wrappers for combined functionality
• Zero dependencies for advanced features (except Redis)

🌟 Features

Core API Support

• ✅ Stock Price Queries: Domestic (KR) and US stocks
• ✅ Stock Information: Company details and market data
• ✅ IPO Schedule: Public offering information and schedules
• ✅ Unified Interface: Query KR/US stocks with fetch_price(symbol, market)
• ✅ Search Functions: Stock search and lookup

Advanced Features (Opt-in)

• 🔧 Env Auto-Detection (v0.9.0): boto3/Twilio-style automatic credential loading
• 🚀 Rate Limiting (v0.8.0): Automatic throttling with logging and monitoring
• 💾 Memory Caching (v0.7.0): Reduce API calls with TTL-based caching
• 🔑 Token Storage (v0.7.0): File or Redis-based token persistence
• 🔄 Wrapper Architecture: Stack features as needed

Technical Features

• 🔧 Context Manager: Automatic resource cleanup
• 🔧 Thread-Safe: All wrappers support concurrent access
• 🔧 Environment Variables: API credentials via env vars
• 🔧 Zero Dependencies: Core wrapper needs only requests and pandas

📦 Installation

pip install korea-investment-stock

Requirements

• Python 3.11 or higher
• Korea Investment Securities API account

🚀 Quick Start

1. Set Up Credentials

# Add to your ~/.zshrc or ~/.bashrc
export KOREA_INVESTMENT_API_KEY="your-api-key"
export KOREA_INVESTMENT_API_SECRET="your-api-secret"
export KOREA_INVESTMENT_ACCOUNT_NO="12345678-01"

2. Basic Usage

from korea_investment_stock import KoreaInvestment

# NEW in v0.9.0: Auto-detect from environment variables!
with KoreaInvestment() as broker:
    result = broker.fetch_price("005930", "KR")

    if result['rt_cd'] == '0':
        price = result['output1']['stck_prpr']
        print(f"Price: {price}원")

Alternative: Explicit parameters (still supported)

from korea_investment_stock import KoreaInvestment
import os

# Explicit parameter passing (backward compatible)
with KoreaInvestment(
    api_key=os.getenv('KOREA_INVESTMENT_API_KEY'),
    api_secret=os.getenv('KOREA_INVESTMENT_API_SECRET'),
    acc_no=os.getenv('KOREA_INVESTMENT_ACCOUNT_NO')
) as broker:
    result = broker.fetch_price("005930", "KR")

Priority: Constructor parameters > Environment variables

# Mixed usage: override specific values
broker = KoreaInvestment(api_key="override-key")  # Others from env vars

📖 API Methods

Stock Price Queries

# Domestic stock price
result = broker.fetch_price("005930", "KR")  # Samsung Electronics

# US stock price (requires real account)
result = broker.fetch_price("AAPL", "US")   # Apple

# Direct methods
result = broker.fetch_domestic_price("J", "005930")
result = broker.fetch_etf_domestic_price("J", "069500")  # KODEX 200
result = broker.fetch_price_detail_oversea("AAPL", "US")

Stock Information

# Stock info
result = broker.fetch_stock_info("005930", "KR")

# Search stock
result = broker.fetch_search_stock_info("005930", "KR")

IPO Schedule

# All IPOs (today + 30 days)
result = broker.fetch_ipo_schedule()

# Specific period
result = broker.fetch_ipo_schedule(
    from_date="20250101",
    to_date="20250131"
)

# Specific symbol
result = broker.fetch_ipo_schedule(symbol="123456")

# Helper methods
status = broker.get_ipo_status(ipo['subscr_dt'])  # "예정", "진행중", "마감"
d_day = broker.calculate_ipo_d_day(ipo['subscr_dt'])  # Days until subscription

Symbol Lists

# KOSPI symbols
result = broker.fetch_kospi_symbols()

# KOSDAQ symbols
result = broker.fetch_kosdaq_symbols()

🔧 Advanced Usage

Multiple Stock Queries

# Query multiple stocks
stocks = [
    ("005930", "KR"),  # Samsung
    ("000660", "KR"),  # SK Hynix
    ("035720", "KR"),  # Kakao
]

results = []
for symbol, market in stocks:
    result = broker.fetch_price(symbol, market)
    results.append(result)
    # Add your own rate limiting here if needed

Mixed KR/US Portfolio

portfolio = [
    ("005930", "KR"),  # Samsung Electronics
    ("AAPL", "US"),    # Apple
    ("035720", "KR"),  # Kakao
    ("MSFT", "US"),    # Microsoft
]

for symbol, market in portfolio:
    result = broker.fetch_price(symbol, market)

    if result['rt_cd'] == '0':
        if market == "KR":
            output = result['output1']
            price = output['stck_prpr']
            print(f"{symbol}: ₩{int(price):,}")
        else:
            output = result['output']
            price = output['last']
            print(f"{symbol}: ${price}")

Error Handling

try:
    result = broker.fetch_price("INVALID", "US")

    if result['rt_cd'] != '0':
        # API returned error
        print(f"API Error: {result['msg1']}")
except ValueError as e:
    # Invalid parameters
    print(f"Invalid request: {e}")
except Exception as e:
    # Network or other errors
    print(f"Error: {e}")

API Rate Limiting (NEW in v0.8.0) 🚀

Problem: Korea Investment API has a 20 calls/second limit. Exceeding this causes errors.

Solution: Automatic rate limiting wrapper that throttles API calls safely.

from korea_investment_stock import KoreaInvestment, RateLimitedKoreaInvestment

# Create base broker
broker = KoreaInvestment(api_key, api_secret, acc_no)

# Wrap with rate limiting (15 calls/second - conservative)
rate_limited = RateLimitedKoreaInvestment(broker, calls_per_second=15)

# Use as normal - rate limiting applied automatically
result = rate_limited.fetch_price("005930", "KR")

# Batch processing is now safe!
for symbol, market in stock_list:
    result = rate_limited.fetch_price(symbol, market)  # Auto-throttled

Configuration Options:

# Production - ultra safe
conservative = RateLimitedKoreaInvestment(broker, calls_per_second=12)

# Testing - closer to limit
aggressive = RateLimitedKoreaInvestment(broker, calls_per_second=18)

# Dynamic adjustment at runtime
rate_limited.adjust_rate_limit(calls_per_second=10)

# Check statistics (NEW: Extended statistics)
stats = rate_limited.get_rate_limit_stats()
print(f"Rate: {stats['calls_per_second']}/sec")
print(f"Total calls: {stats['total_calls']}")
print(f"Throttled: {stats['throttled_calls']} ({stats['throttle_rate']*100:.1f}%)")
print(f"Total wait time: {stats['total_wait_time']:.2f}s")
print(f"Avg wait time: {stats['avg_wait_time']:.3f}s")

Logging & Monitoring (NEW):

Built-in logging and extended statistics for production monitoring:

import logging
from korea_investment_stock import KoreaInvestment, RateLimitedKoreaInvestment

# Enable DEBUG logging to see throttle events
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

broker = KoreaInvestment(api_key, api_secret, acc_no)
rate_limited = RateLimitedKoreaInvestment(broker, calls_per_second=15)

# Each throttle event will be logged:
# "Rate limit: waiting 0.067s (call #2)"
result = rate_limited.fetch_price("005930", "KR")

# Get detailed statistics after batch processing
stats = rate_limited.get_rate_limit_stats()
print(f"Performance Metrics:")
print(f"  Total calls: {stats['total_calls']}")
print(f"  Throttled: {stats['throttled_calls']} ({stats['throttle_rate']*100:.1f}%)")
print(f"  Total wait time: {stats['total_wait_time']:.2f}s")
print(f"  Avg wait time: {stats['avg_wait_time']:.3f}s")

Extended Statistics Fields:

{
    'calls_per_second': 15.0,       # Configured rate limit
    'total_calls': 500,              # Total API calls made
    'throttled_calls': 450,          # Calls that were throttled (NEW)
    'throttle_rate': 0.90,           # Throttle percentage (NEW)
    'total_wait_time': 28.5,         # Total time spent waiting (NEW)
    'avg_wait_time': 0.0633          # Average wait per throttle (NEW)
}

Combined with Cache (Recommended!):

from korea_investment_stock import (
    KoreaInvestment,
    CachedKoreaInvestment,
    RateLimitedKoreaInvestment
)

# Stack both wrappers for maximum efficiency
broker = KoreaInvestment(api_key, api_secret, acc_no)
cached = CachedKoreaInvestment(broker, price_ttl=5)
safe_broker = RateLimitedKoreaInvestment(cached, calls_per_second=15)

# Benefits:
# ✅ Cache reduces API calls (performance)
# ✅ Rate limit protects cache misses (safety)
# ✅ No API errors, maximum efficiency

Performance Impact:

Scenario	Without Rate Limit	With Rate Limit (15/sec)
10 API calls	~1-3 sec	~0.67 sec
100 API calls	~10-30 sec	~6.7 sec
500 API calls	FAILS (rate limit errors)	~33 sec (100% success)

Features:

• 🔒 Thread-safe with threading.Lock
• ⚡ Token bucket algorithm
• 📊 Real-time statistics with extended metrics (NEW)
• 📝 Built-in logging for throttle events (NEW)
• 🎛️ Dynamic rate adjustment
• 🔧 Context manager support

See: examples/stress_test.py for 500+ API calls example with logging and statistics

---

Memory Caching (Optional)

Reduce API calls and improve response times with built-in memory caching:

from korea_investment_stock import KoreaInvestment, CachedKoreaInvestment

# Create base broker
broker = KoreaInvestment(api_key, api_secret, acc_no)

# Wrap with caching (opt-in)
cached_broker = CachedKoreaInvestment(broker, price_ttl=5)

# First call: API request (cache miss)
result1 = cached_broker.fetch_price("005930", "KR")  # ~200ms

# Second call: from cache (cache hit)
result2 = cached_broker.fetch_price("005930", "KR")  # <1ms

# Cache statistics
stats = cached_broker.get_cache_stats()
print(f"Hit rate: {stats['hit_rate']}")  # "50.00%"

TTL Configuration (in seconds):

cached_broker = CachedKoreaInvestment(
    broker,
    price_ttl=5,        # Real-time price: 5 seconds
    stock_info_ttl=300, # Stock info: 5 minutes
    symbols_ttl=3600,   # Symbol lists: 1 hour
    ipo_ttl=1800        # IPO schedule: 30 minutes
)

Performance Benefits:

• 📉 API calls reduced by 30-50%
• ⚡ Response time improved by 90%+ (cached queries)
• 🔒 Thread-safe with automatic expiration
• 💾 No external dependencies (memory-only)

See: examples/cached_basic_example.py for comprehensive examples

---

Token Storage (Advanced)

Persist API tokens across sessions for faster initialization:

File-based Storage (Default):

from korea_investment_stock import KoreaInvestment, FileTokenStorage
from pathlib import Path

# Default: ~/.cache/kis/token.key
broker = KoreaInvestment(
    api_key=api_key,
    api_secret=api_secret,
    acc_no=acc_no,
    token_storage=FileTokenStorage()
)

# Custom path
custom_storage = FileTokenStorage(Path("/custom/path/token.key"))
broker = KoreaInvestment(api_key, api_secret, acc_no, token_storage=custom_storage)

Redis-based Storage (Distributed):

from korea_investment_stock import KoreaInvestment, RedisTokenStorage

# Install redis support: pip install korea-investment-stock[redis]
redis_storage = RedisTokenStorage(
    host='localhost',
    port=6379,
    db=0,
    password=None  # Optional
)

broker = KoreaInvestment(
    api_key=api_key,
    api_secret=api_secret,
    acc_no=acc_no,
    token_storage=redis_storage
)

Benefits:

• ⚡ Skip token generation on subsequent runs (24h TTL)
• 🔄 Share tokens across multiple processes (Redis)
• 💾 Automatic token expiration and renewal
• 🔒 Thread-safe token management

See: examples/redis_token_example.py for Redis setup

📊 Response Format

Domestic Stock (KR)

{
    'rt_cd': '0',               # Return code ('0' = success)
    'msg1': '정상처리되었습니다',   # Message
    'output1': {
        'stck_prpr': '62600',   # Current price
        'prdy_vrss': '1600',    # Change from previous day
        'prdy_ctrt': '2.62',    # Change rate (%)
        'stck_oprc': '61000',   # Opening price
        'stck_hgpr': '63000',   # High price
        'stck_lwpr': '60500',   # Low price
        'acml_vol': '15234567', # Volume (거래량)
        'hts_avls': '3735468'   # Market cap (시가총액, 억원)
        # ... more fields
    }
}

US Stock (US)

{
    'rt_cd': '0',
    'msg1': '정상처리되었습니다',
    'output': {
        'rsym': 'DNASAAPL',        # Exchange + Symbol
        'last': '211.16',          # Current price
        'open': '210.56',          # Opening price
        'high': '212.13',          # High price
        'low': '209.86',           # Low price
        'tvol': '39765812',        # Volume (거래량)
        'tomv': '3250000000000',   # Market cap (시가총액)
        'shar': '15384171000',     # Shares outstanding (상장주수)
        't_xdif': '1.72',          # Change
        't_xrat': '-0.59',         # Change rate (%)
        'perx': '32.95',           # PER
        'pbrx': '47.23',           # PBR
        'epsx': '6.41',            # EPS
        'bpsx': '4.47'             # BPS
        # ... more fields
    }
}

⚠️ Important Notes

API Rate Limits

• Korea Investment API: 20 requests/second limit
• NEW in v0.8.0: Built-in RateLimitedKoreaInvestment wrapper (recommended!)
• Without rate limiting: API errors when limit exceeded
• Best practice: Use RateLimitedKoreaInvestment or implement your own

US Stocks

• Auto-detects exchange (NASDAQ → NYSE → AMEX)
• Includes financial ratios (PER, PBR, EPS, BPS)
• Requires real account (paper trading not supported)

Context Manager

Always use context manager for proper resource cleanup:

# ✅ Good: Automatic cleanup
with KoreaInvestment(api_key, api_secret, acc_no) as broker:
    result = broker.fetch_price("005930", "KR")

# ❌ Bad: Manual cleanup required
broker = KoreaInvestment(api_key, api_secret, acc_no)
result = broker.fetch_price("005930", "KR")
broker.shutdown()  # Must call manually

Optional Features

All advanced features are opt-in wrappers:

• CachedKoreaInvestment: Memory caching (v0.7.0)
• RateLimitedKoreaInvestment: Rate limiting (v0.8.0)
• FileTokenStorage / RedisTokenStorage: Token persistence (v0.7.0)

You control what features to use based on your needs.

🔨 Best Practices & Recommended Setup

Production Setup (Recommended)

Combine all features for optimal performance and safety:

from korea_investment_stock import (
    KoreaInvestment,
    CachedKoreaInvestment,
    RateLimitedKoreaInvestment,
    FileTokenStorage  # or RedisTokenStorage for distributed
)

# 1. Core broker with token persistence
broker = KoreaInvestment(
    api_key=api_key,
    api_secret=api_secret,
    acc_no=acc_no,
    token_storage=FileTokenStorage()  # Skip token generation on restart
)

# 2. Add caching layer (reduce API calls)
cached = CachedKoreaInvestment(broker, price_ttl=5)

# 3. Add rate limiting (prevent API errors)
safe_broker = RateLimitedKoreaInvestment(cached, calls_per_second=15)

# 4. Use with context manager
with safe_broker:
    for symbol, market in portfolio:
        result = safe_broker.fetch_price(symbol, market)
        # ✅ Cached: <1ms response
        # ✅ Rate limited: No API errors
        # ✅ Token persisted: Fast restarts

Custom Implementation Examples

If you prefer to implement your own features:

Custom Rate Limiter:

import time

class CustomRateLimiter:
    def __init__(self, calls_per_second=15):
        self.min_interval = 1.0 / calls_per_second
        self.last_call = 0

    def wait(self):
        elapsed = time.time() - self.last_call
        if elapsed < self.min_interval:
            time.sleep(self.min_interval - elapsed)
        self.last_call = time.time()

limiter = CustomRateLimiter(calls_per_second=15)
for symbol, market in stocks:
    limiter.wait()
    result = broker.fetch_price(symbol, market)

Custom Caching:

from datetime import datetime, timedelta

class CustomCache:
    def __init__(self, broker, ttl_minutes=5):
        self.broker = broker
        self.cache = {}
        self.ttl = timedelta(minutes=ttl_minutes)

    def fetch_price_cached(self, symbol, market):
        key = f"{symbol}:{market}"
        now = datetime.now()

        if key in self.cache:
            cached_time, result = self.cache[key]
            if now - cached_time < self.ttl:
                return result

        result = self.broker.fetch_price(symbol, market)
        self.cache[key] = (now, result)
        return result

Custom Retry Logic:

import time

def fetch_with_retry(broker, symbol, market, max_retries=3):
    for attempt in range(max_retries):
        try:
            result = broker.fetch_price(symbol, market)
            if result['rt_cd'] == '0':
                return result
            time.sleep(2 ** attempt)  # Exponential backoff
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)

📚 Examples

See the examples/ directory for complete working examples:

Example	Description	Features
basic_example.py	Simple usage patterns	Basic API calls, error handling
cached_basic_example.py	Caching examples	Memory cache, TTL configuration, statistics
stress_test.py	Batch processing (500 calls)	Rate limiting, batch operations, performance
us_stock_price_example.py	US stock queries	Exchange detection, financial ratios
redis_token_example.py	Token persistence	Redis setup, distributed tokens

Quick Start:

# Set environment variables first
export KOREA_INVESTMENT_API_KEY="your-key"
export KOREA_INVESTMENT_API_SECRET="your-secret"
export KOREA_INVESTMENT_ACCOUNT_NO="12345678-01"

# Run examples
python examples/basic_example.py
python examples/stress_test.py

🔄 Migration Guide

v0.9.0 Changes (Non-breaking)

Environment variable auto-detection (like boto3/Twilio):

# Before v0.9.0 - Required explicit env var reading
import os
broker = KoreaInvestment(
    api_key=os.getenv('KOREA_INVESTMENT_API_KEY'),
    api_secret=os.getenv('KOREA_INVESTMENT_API_SECRET'),
    acc_no=os.getenv('KOREA_INVESTMENT_ACCOUNT_NO')
)

# v0.9.0+ - Auto-detect from environment variables
broker = KoreaInvestment()  # Reads KOREA_INVESTMENT_* env vars automatically

# Mixed usage - override specific values
broker = KoreaInvestment(api_key="override-key")  # Others from env vars

Priority: Constructor parameters > Environment variables

No breaking changes - all existing code continues to work

---

v0.8.0 Changes (Breaking)

Mock mode removed (See CHANGELOG.md for details):

# v0.7.x (Old)
broker = KoreaInvestment(api_key, api_secret, acc_no, mock=True)

# v0.8.0 (New)
broker = KoreaInvestment(api_key, api_secret, acc_no)

Rate limiting added (Opt-in):

# NEW in v0.8.0: Built-in rate limiting
from korea_investment_stock import RateLimitedKoreaInvestment

rate_limited = RateLimitedKoreaInvestment(broker, calls_per_second=15)

v0.7.0 Changes (Non-breaking)

Added features (all opt-in):

• Memory caching: CachedKoreaInvestment
• Token storage: FileTokenStorage, RedisTokenStorage
• Package restructure: Feature-based modules

No breaking changes - all v0.6.0 code works in v0.7.0

v0.6.0 Changes (Breaking)

Removed features (~6,000 lines of code):

• Built-in rate limiting (now available as opt-in in v0.8.0)
• Built-in caching (now available as opt-in in v0.7.0)
• Batch processing methods
• Monitoring and statistics
• Visualization tools
• Automatic retry decorators

API changes:

# v0.5.0 (Old)
results = broker.fetch_price_list([("005930", "KR"), ("AAPL", "US")])

# v0.6.0+ (New)
results = []
for symbol, market in [("005930", "KR"), ("AAPL", "US")]:
    result = broker.fetch_price(symbol, market)
    results.append(result)

See CHANGELOG.md for complete migration guide.

📖 Documentation

• Official API Docs
• GitHub Issues

🧪 Testing

Test Categories

This project uses pytest markers to categorize tests:

Marker	Description	Docker Required
unit	Fast unit tests with mocks	No
integration	Tests with real Docker containers	Yes

Running Tests

# Install dev dependencies
pip install -e ".[dev,redis]"

# Run unit tests only (fast, no Docker)
pytest -m "not integration"

# Run integration tests only (requires Docker)
pytest -m integration

# Run all tests
pytest

# Run with verbose output
pytest -v

Integration Tests (Testcontainers)

Integration tests use testcontainers to spin up real Docker containers:

# Redis integration tests
pytest -m integration -v

# Example output:
# test_redis_integration.py::TestRedisTokenStorageIntegration::test_save_and_load PASSED
# test_redis_integration.py::TestRedisTokenStorageIntegration::test_connection_pool PASSED
# test_redis_integration.py::TestRedisTokenStorageIntegration::test_ttl_actual_expiry PASSED
# ...

Features:

• Real Redis container for token storage tests
• Automatic container lifecycle management
• Graceful skip when Docker is unavailable
• Thread-safe connection pool testing
• Actual TTL expiry verification

🤝 Contributing

Contributions welcome! Please:

1. Fork the repository
2. Create a feature branch
3. Submit a pull request

📝 License

MIT License - see LICENSE file

⚡ Performance Tips

Quick Wins

1. Use RateLimitedKoreaInvestment: Prevent API errors (v0.8.0)
2. Use CachedKoreaInvestment: Reduce redundant API calls (v0.7.0)
3. Use FileTokenStorage: Skip token generation on restart (v0.7.0)
4. Stack all wrappers: Maximum performance and safety
5. Use context manager: Ensure proper cleanup

Performance Comparison

Setup	100 stocks (same)	100 stocks (unique)	Token generation
Basic	~33 sec	~33 sec	Every run (~1 sec)
+ Rate Limit	~6.7 sec	~6.7 sec	Every run (~1 sec)
+ Cache	~0.1 sec	~6.7 sec	Every run (~1 sec)
+ Token Storage	~0.1 sec	~6.7 sec	First run only
All combined	~0.1 sec	~6.7 sec	First run only

Optimization Strategy

# Choose based on your use case:

# Real-time trading (fast changing prices)
cached = CachedKoreaInvestment(broker, price_ttl=1)

# Analysis/backtesting (less frequent updates)
cached = CachedKoreaInvestment(broker, price_ttl=60)

# Production batch processing
safe_broker = RateLimitedKoreaInvestment(
    CachedKoreaInvestment(broker, price_ttl=5),
    calls_per_second=15
)

# Distributed systems
broker = KoreaInvestment(
    api_key, api_secret, acc_no,
    token_storage=RedisTokenStorage()  # Share tokens across instances
)

🙏 Credits

• Korea Investment Securities for providing the OpenAPI
• Original contributors: Jonghun Yoo, Brayden Jo, Frank Oh

---

Remember: This is a pure wrapper. You control rate limiting, caching, error handling, and monitoring according to your needs.