Compare commits

..

12 Commits

Author SHA1 Message Date
agentson
cb2e3fae57 fix: reduce rate limit from 10 to 5 RPS to avoid API errors (issue #43)
Some checks failed
CI / test (pull_request) Has been cancelled
Reduce RATE_LIMIT_RPS from 10.0 to 5.0 to prevent "초당 거래건수를
초과하였습니다" (EGW00201) errors from KIS API.

Docker logs showed this was the most frequent error (70% of failures),
occurring when multiple stocks are scanned rapidly.

Changes:
- src/config.py: RATE_LIMIT_RPS 10.0 → 5.0
- .env.example: Update default and add explanation comment

Trade-off: Slower API throughput, but more reliable operation.
Can be tuned per deployment via environment variable.

Fixes: #43

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-02-05 00:12:57 +09:00
5e4c68c9d8 Merge pull request 'fix: add token refresh lock to prevent concurrent API calls (issue #42)' (#46) from feature/issue-42-token-refresh-lock into main
Some checks failed
CI / test (push) Has been cancelled
Reviewed-on: #46
2026-02-05 00:11:04 +09:00
agentson
95f540e5df fix: add token refresh lock to prevent concurrent API calls (issue #42)
Some checks failed
CI / test (pull_request) Has been cancelled
Add asyncio.Lock to prevent multiple coroutines from simultaneously
refreshing the KIS access token, which hits the 1-per-minute rate
limit (EGW00133: "접근토큰 발급 잠시 후 다시 시도하세요").

Changes:
- Add self._token_lock in KISBroker.__init__
- Wrap token refresh in async with self._token_lock
- Re-check token validity after acquiring lock (double-check pattern)
- Add concurrent token refresh test (5 parallel requests → 1 API call)

The lock ensures that when multiple coroutines detect an expired token,
only the first one refreshes while others wait and reuse the result.

Fixes: #42

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-02-05 00:08:56 +09:00
0087a6b20a Merge pull request 'fix: handle dict and list formats in overseas balance output2 (issue #41)' (#45) from feature/issue-41-keyerror-balance into main
Some checks failed
CI / test (push) Has been cancelled
Reviewed-on: #45
2026-02-05 00:06:25 +09:00
agentson
3dfd7c0935 fix: handle dict and list formats in overseas balance output2 (issue #41)
Some checks failed
CI / test (pull_request) Has been cancelled
Add type checking for output2 response from get_overseas_balance API.
The API can return either list format [{}] or dict format {}, causing
KeyError when accessing output2[0].

Changes:
- Check isinstance before accessing output2[0]
- Handle list, dict, and empty cases
- Add safe fallback with "or" for empty strings
- Add 3 test cases for list/dict/empty formats

Fixes: #41

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-02-05 00:04:36 +09:00
4b2bb25d03 Merge pull request 'docs: add Telegram notifications documentation (issue #35)' (#40) from feature/issue-35-telegram-docs into main
Some checks failed
CI / test (push) Has been cancelled
Reviewed-on: #40
2026-02-04 23:49:45 +09:00
agentson
881bbb4240 docs: add Telegram notifications documentation (issue #35)
Some checks failed
CI / test (pull_request) Has been cancelled
Update project documentation to include Telegram notification feature
that was added in issues #31-34.

Changes:
- CLAUDE.md: Add Telegram quick setup section with examples
- README.md (Korean): Add 텔레그램 알림 section with setup guide
- docs/architecture.md: Add Notifications component documentation
  - New section explaining TelegramClient architecture
  - Add notification step to data flow diagram
  - Add Telegram config to environment variables
  - Document error handling for notification failures

Documentation covers:
- Quick setup instructions (bot creation, chat ID, env config)
- Notification types (trades, circuit breaker, fat-finger, etc.)
- Fail-safe behavior (notifications never crash trading)
- Links to detailed guide in src/notifications/README.md

Project structure updated to reflect notifications/ directory and
updated test count (273 tests).

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-02-04 23:48:01 +09:00
5f7d61748b Merge pull request 'feat: integrate TelegramClient into main trading loop (issue #34)' (#39) from feature/issue-34-main-integration into main
Some checks failed
CI / test (push) Has been cancelled
Reviewed-on: #39
2026-02-04 23:44:49 +09:00
agentson
972e71a2f1 feat: integrate TelegramClient into main trading loop (issue #34)
Some checks failed
CI / test (pull_request) Has been cancelled
Integrate Telegram notifications throughout the main trading loop to provide
real-time alerts for critical events and trading activities.

Changes:
- Add TelegramClient initialization in run() function
- Send system startup notification on agent start
- Send market open/close notifications when markets change state
- Send trade execution notifications for BUY/SELL orders
- Send fat finger rejection notifications when orders are blocked
- Send circuit breaker notifications when loss threshold is exceeded
- Pass telegram client to trading_cycle() function
- Add tests for all notification scenarios in test_main.py

All notifications wrapped in try/except to ensure trading continues even
if Telegram API fails. Notifications are non-blocking and do not affect
core trading logic.

Test coverage: 273 tests passed, overall coverage 79%

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-02-04 23:42:31 +09:00
614b9939b1 Merge pull request 'feat: add Telegram configuration to settings (issue #33)' (#38) from feature/issue-33-telegram-config into main
Some checks failed
CI / test (push) Has been cancelled
Reviewed-on: #38
2026-02-04 21:42:49 +09:00
agentson
6dbc2afbf4 feat: add Telegram configuration to settings (issue #33)
Some checks failed
CI / test (pull_request) Has been cancelled
Add Telegram notification configuration:
- src/config.py: Add TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID, TELEGRAM_ENABLED
- .env.example: Add Telegram section with setup instructions

Fields added after S3_REGION (line 55).
Follows existing optional API pattern (NEWS_API_KEY, etc.).
No breaking changes to existing settings.

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-02-04 21:34:05 +09:00
6c96f9ac64 Merge pull request 'test: add comprehensive TelegramClient tests (issue #32)' (#37) from feature/issue-32-telegram-tests into main
Some checks failed
CI / test (push) Has been cancelled
Reviewed-on: #37
2026-02-04 21:33:19 +09:00
9 changed files with 853 additions and 42 deletions

View File

@@ -16,8 +16,9 @@ CONFIDENCE_THRESHOLD=80
# Database
DB_PATH=data/trade_logs.db
# Rate Limiting
RATE_LIMIT_RPS=10.0
# Rate Limiting (requests per second for KIS API)
# Reduced to 5.0 to avoid "초당 거래건수 초과" errors (EGW00201)
RATE_LIMIT_RPS=5.0
# Trading Mode (paper / live)
MODE=paper
@@ -26,3 +27,10 @@ MODE=paper
# NEWS_API_KEY=your_news_api_key_here
# NEWS_API_PROVIDER=alphavantage
# MARKET_DATA_API_KEY=your_market_data_key_here
# Telegram Notifications (optional)
# Get bot token from @BotFather on Telegram
# Get chat ID from @userinfobot or your chat
# TELEGRAM_BOT_TOKEN=1234567890:ABCdefGHIjklMNOpqrsTUVwxyz
# TELEGRAM_CHAT_ID=123456789
# TELEGRAM_ENABLED=true

View File

@@ -17,6 +17,34 @@ pytest -v --cov=src
python -m src.main --mode=paper
```
## Telegram Notifications (Optional)
Get real-time alerts for trades, circuit breakers, and system events via Telegram.
### Quick Setup
1. **Create bot**: Message [@BotFather](https://t.me/BotFather) on Telegram → `/newbot`
2. **Get chat ID**: Message [@userinfobot](https://t.me/userinfobot) → `/start`
3. **Configure**: Add to `.env`:
```bash
TELEGRAM_BOT_TOKEN=1234567890:ABCdefGHIjklMNOpqrsTUVwxyz
TELEGRAM_CHAT_ID=123456789
TELEGRAM_ENABLED=true
```
4. **Test**: Start bot conversation (`/start`), then run the agent
**Full documentation**: [src/notifications/README.md](src/notifications/README.md)
### What You'll Get
- 🟢 Trade execution alerts (BUY/SELL with confidence)
- 🚨 Circuit breaker trips (automatic trading halt)
- ⚠️ Fat-finger rejections (oversized orders blocked)
- Market open/close notifications
- 📝 System startup/shutdown status
**Fail-safe**: Notifications never crash the trading system. Missing credentials or API errors are logged but trading continues normally.
## Documentation
- **[Workflow Guide](docs/workflow.md)** — Git workflow policy and agent-based development
@@ -42,11 +70,12 @@ src/
├── core/ # Risk manager (READ-ONLY)
├── evolution/ # Self-improvement optimizer
├── markets/ # Market schedules and timezone handling
├── notifications/ # Telegram real-time alerts
├── db.py # SQLite trade logging
├── main.py # Trading loop orchestrator
└── config.py # Settings (from .env)
tests/ # 54 tests across 4 files
tests/ # 273 tests across 13 files
docs/ # Extended documentation
```

View File

@@ -29,6 +29,7 @@ KIS(한국투자증권) API로 매매하고, Google Gemini로 판단하며, 자
| 브로커 | `src/broker/kis_api.py` | KIS API 비동기 래퍼 (토큰 갱신, 레이트 리미터, 해시키) |
| 두뇌 | `src/brain/gemini_client.py` | Gemini 프롬프트 구성 및 JSON 응답 파싱 |
| 방패 | `src/core/risk_manager.py` | 서킷 브레이커 + 팻 핑거 체크 |
| 알림 | `src/notifications/telegram_client.py` | 텔레그램 실시간 거래 알림 (선택사항) |
| 진화 | `src/evolution/optimizer.py` | 실패 패턴 분석 → 새 전략 생성 → 테스트 → PR |
| DB | `src/db.py` | SQLite 거래 로그 기록 |
@@ -75,6 +76,34 @@ python -m src.main --mode=paper
docker compose up -d ouroboros
```
## 텔레그램 알림 (선택사항)
거래 실행, 서킷 브레이커 발동, 시스템 상태 등을 텔레그램으로 실시간 알림 받을 수 있습니다.
### 빠른 설정
1. **봇 생성**: 텔레그램에서 [@BotFather](https://t.me/BotFather) 메시지 → `/newbot` 명령
2. **채팅 ID 확인**: [@userinfobot](https://t.me/userinfobot) 메시지 → `/start` 명령
3. **환경변수 설정**: `.env` 파일에 추가
```bash
TELEGRAM_BOT_TOKEN=1234567890:ABCdefGHIjklMNOpqrsTUVwxyz
TELEGRAM_CHAT_ID=123456789
TELEGRAM_ENABLED=true
```
4. **테스트**: 봇과 대화 시작 (`/start` 전송) 후 에이전트 실행
**상세 문서**: [src/notifications/README.md](src/notifications/README.md)
### 알림 종류
- 🟢 거래 체결 알림 (BUY/SELL + 신뢰도)
- 🚨 서킷 브레이커 발동 (자동 거래 중단)
- ⚠️ 팻 핑거 차단 (과도한 주문 차단)
- 장 시작/종료 알림
- 📝 시스템 시작/종료 상태
**안전장치**: 알림 실패해도 거래는 계속 진행됩니다. 텔레그램 API 오류나 설정 누락이 있어도 거래 시스템은 정상 작동합니다.
## 테스트
35개 테스트가 TDD 방식으로 구현 전에 먼저 작성되었습니다.
@@ -104,15 +133,16 @@ The-Ouroboros/
│ ├── agents.md # AI 에이전트 페르소나 정의
│ └── skills.md # 사용 가능한 도구 목록
├── src/
│ ├── config.py # Pydantic 설정
│ ├── logging_config.py # JSON 구조화 로깅
│ ├── db.py # SQLite 거래 기록
│ ├── main.py # 비동기 거래 루프
│ ├── broker/kis_api.py # KIS API 클라이언트
│ ├── brain/gemini_client.py # Gemini 의사결정 엔진
│ ├── core/risk_manager.py # 리스크 관리
│ ├── evolution/optimizer.py # 전략 진화 엔진
── strategies/base.py # 전략 베이스 클래스
│ ├── config.py # Pydantic 설정
│ ├── logging_config.py # JSON 구조화 로깅
│ ├── db.py # SQLite 거래 기록
│ ├── main.py # 비동기 거래 루프
│ ├── broker/kis_api.py # KIS API 클라이언트
│ ├── brain/gemini_client.py # Gemini 의사결정 엔진
│ ├── core/risk_manager.py # 리스크 관리
│ ├── notifications/telegram_client.py # 텔레그램 알림
── evolution/optimizer.py # 전략 진화 엔진
│ └── strategies/base.py # 전략 베이스 클래스
├── tests/ # TDD 테스트 스위트
├── Dockerfile # 멀티스테이지 빌드
├── docker-compose.yml # 서비스 오케스트레이션

View File

@@ -51,7 +51,26 @@ Self-evolving AI trading agent for global stock markets via KIS (Korea Investmen
- **Fat-Finger Protection**: Rejects orders exceeding 30% of available cash
- Must always be enforced, cannot be disabled
### 4. Evolution (`src/evolution/optimizer.py`)
### 4. Notifications (`src/notifications/telegram_client.py`)
**TelegramClient** — Real-time event notifications via Telegram Bot API
- Sends alerts for trades, circuit breakers, fat-finger rejections, system events
- Non-blocking: failures are logged but never crash trading
- Rate-limited: 1 message/second default to respect Telegram API limits
- Auto-disabled when credentials missing
- Gracefully handles API errors, network timeouts, invalid tokens
**Notification Types:**
- Trade execution (BUY/SELL with confidence)
- Circuit breaker trips (critical alert)
- Fat-finger protection triggers (order rejection)
- Market open/close events
- System startup/shutdown status
**Setup:** See [src/notifications/README.md](../src/notifications/README.md) for bot creation and configuration.
### 5. Evolution (`src/evolution/optimizer.py`)
**StrategyOptimizer** — Self-improvement loop
@@ -115,6 +134,14 @@ Self-evolving AI trading agent for global stock markets via KIS (Korea Investmen
┌──────────────────────────────────┐
│ Notifications: Send Alert │
│ - Trade execution notification │
│ - Non-blocking (errors logged) │
│ - Rate-limited to 1/sec │
└──────────────────┬────────────────┘
┌──────────────────────────────────┐
│ Database: Log Trade │
│ - SQLite (data/trades.db) │
│ - Track: action, confidence, │
@@ -164,6 +191,11 @@ CONFIDENCE_THRESHOLD=80
MAX_LOSS_PCT=3.0
MAX_ORDER_PCT=30.0
ENABLED_MARKETS=KR,US_NASDAQ # Comma-separated market codes
# Telegram Notifications (optional)
TELEGRAM_BOT_TOKEN=1234567890:ABCdefGHIjklMNOpqrsTUVwxyz
TELEGRAM_CHAT_ID=123456789
TELEGRAM_ENABLED=true
```
Tests use in-memory SQLite (`DB_PATH=":memory:"`) and dummy credentials via `tests/conftest.py`.
@@ -189,3 +221,12 @@ Tests use in-memory SQLite (`DB_PATH=":memory:"`) and dummy credentials via `tes
- Wait until next market opens
- Use `get_next_market_open()` to calculate wait time
- Sleep until market open time
### Telegram API Errors
- Log warning but continue trading
- Missing credentials → auto-disable notifications
- Network timeout → skip notification, no retry
- Invalid token → log error, trading unaffected
- Rate limit exceeded → queued via rate limiter
**Guarantee**: Notification failures never interrupt trading operations.

View File

@@ -55,6 +55,7 @@ class KISBroker:
self._session: aiohttp.ClientSession | None = None
self._access_token: str | None = None
self._token_expires_at: float = 0.0
self._token_lock = asyncio.Lock()
self._rate_limiter = LeakyBucket(settings.RATE_LIMIT_RPS)
def _get_session(self) -> aiohttp.ClientSession:
@@ -80,30 +81,42 @@ class KISBroker:
# ------------------------------------------------------------------
async def _ensure_token(self) -> str:
"""Return a valid access token, refreshing if expired."""
"""Return a valid access token, refreshing if expired.
Uses a lock to prevent concurrent token refresh attempts that would
hit the API's 1-per-minute rate limit (EGW00133).
"""
# Fast path: check without lock
now = asyncio.get_event_loop().time()
if self._access_token and now < self._token_expires_at:
return self._access_token
logger.info("Refreshing KIS access token")
session = self._get_session()
url = f"{self._base_url}/oauth2/tokenP"
body = {
"grant_type": "client_credentials",
"appkey": self._app_key,
"appsecret": self._app_secret,
}
# Slow path: acquire lock and refresh
async with self._token_lock:
# Re-check after acquiring lock (another coroutine may have refreshed)
now = asyncio.get_event_loop().time()
if self._access_token and now < self._token_expires_at:
return self._access_token
async with session.post(url, json=body) as resp:
if resp.status != 200:
text = await resp.text()
raise ConnectionError(f"Token refresh failed ({resp.status}): {text}")
data = await resp.json()
logger.info("Refreshing KIS access token")
session = self._get_session()
url = f"{self._base_url}/oauth2/tokenP"
body = {
"grant_type": "client_credentials",
"appkey": self._app_key,
"appsecret": self._app_secret,
}
self._access_token = data["access_token"]
self._token_expires_at = now + data.get("expires_in", 86400) - 60 # 1-min buffer
logger.info("Token refreshed successfully")
return self._access_token
async with session.post(url, json=body) as resp:
if resp.status != 200:
text = await resp.text()
raise ConnectionError(f"Token refresh failed ({resp.status}): {text}")
data = await resp.json()
self._access_token = data["access_token"]
self._token_expires_at = now + data.get("expires_in", 86400) - 60 # 1-min buffer
logger.info("Token refreshed successfully")
return self._access_token
# ------------------------------------------------------------------
# Hash Key (required for POST bodies)

View File

@@ -37,7 +37,8 @@ class Settings(BaseSettings):
DB_PATH: str = "data/trade_logs.db"
# Rate Limiting (requests per second for KIS API)
RATE_LIMIT_RPS: float = 10.0
# Reduced to 5.0 to avoid EGW00201 "초당 거래건수 초과" errors
RATE_LIMIT_RPS: float = 5.0
# Trading mode
MODE: str = Field(default="paper", pattern="^(paper|live)$")
@@ -54,6 +55,11 @@ class Settings(BaseSettings):
S3_BUCKET_NAME: str | None = None
S3_REGION: str = "us-east-1"
# Telegram Notifications (optional)
TELEGRAM_BOT_TOKEN: str | None = None
TELEGRAM_CHAT_ID: str | None = None
TELEGRAM_ENABLED: bool = True
model_config = {"env_file": ".env", "env_file_encoding": "utf-8"}
@property

View File

@@ -10,6 +10,7 @@ import argparse
import asyncio
import logging
import signal
import sys
from datetime import UTC, datetime
from typing import Any
@@ -23,11 +24,12 @@ from src.context.layer import ContextLayer
from src.context.store import ContextStore
from src.core.criticality import CriticalityAssessor
from src.core.priority_queue import PriorityTaskQueue
from src.core.risk_manager import CircuitBreakerTripped, RiskManager
from src.core.risk_manager import CircuitBreakerTripped, FatFingerRejected, RiskManager
from src.db import init_db, log_trade
from src.logging.decision_logger import DecisionLogger
from src.logging_config import setup_logging
from src.markets.schedule import MarketInfo, get_next_market_open, get_open_markets
from src.notifications.telegram_client import TelegramClient
logger = logging.getLogger(__name__)
@@ -62,6 +64,7 @@ async def trading_cycle(
decision_logger: DecisionLogger,
context_store: ContextStore,
criticality_assessor: CriticalityAssessor,
telegram: TelegramClient,
market: MarketInfo,
stock_code: str,
) -> None:
@@ -92,9 +95,17 @@ async def trading_cycle(
balance_data = await overseas_broker.get_overseas_balance(market.exchange_code)
output2 = balance_data.get("output2", [{}])
total_eval = float(output2[0].get("frcr_evlu_tota", "0")) if output2 else 0
total_cash = float(output2[0].get("frcr_dncl_amt_2", "0")) if output2 else 0
purchase_total = float(output2[0].get("frcr_buy_amt_smtl", "0")) if output2 else 0
# Handle both list and dict response formats
if isinstance(output2, list) and output2:
balance_info = output2[0]
elif isinstance(output2, dict):
balance_info = output2
else:
balance_info = {}
total_eval = float(balance_info.get("frcr_evlu_tota", "0") or "0")
total_cash = float(balance_info.get("frcr_dncl_amt_2", "0") or "0")
purchase_total = float(balance_info.get("frcr_buy_amt_smtl", "0") or "0")
current_price = float(price_data.get("output", {}).get("last", "0"))
foreigner_net = 0.0 # Not available for overseas
@@ -199,11 +210,23 @@ async def trading_cycle(
order_amount = current_price * quantity
# 4. Risk check BEFORE order
risk.validate_order(
current_pnl_pct=pnl_pct,
order_amount=order_amount,
total_cash=total_cash,
)
try:
risk.validate_order(
current_pnl_pct=pnl_pct,
order_amount=order_amount,
total_cash=total_cash,
)
except FatFingerRejected as exc:
try:
await telegram.notify_fat_finger(
stock_code=stock_code,
order_amount=exc.order_amount,
total_cash=exc.total_cash,
max_pct=exc.max_pct,
)
except Exception as notify_exc:
logger.warning("Fat finger notification failed: %s", notify_exc)
raise # Re-raise to prevent trade
# 5. Send order
if market.is_domestic:
@@ -223,6 +246,19 @@ async def trading_cycle(
)
logger.info("Order result: %s", result.get("msg1", "OK"))
# 5.5. Notify trade execution
try:
await telegram.notify_trade_execution(
stock_code=stock_code,
market=market.name,
action=decision.action,
quantity=quantity,
price=current_price,
confidence=decision.confidence,
)
except Exception as exc:
logger.warning("Telegram notification failed: %s", exc)
# 6. Log trade
log_trade(
conn=db_conn,
@@ -266,6 +302,13 @@ async def run(settings: Settings) -> None:
decision_logger = DecisionLogger(db_conn)
context_store = ContextStore(db_conn)
# Initialize Telegram notifications
telegram = TelegramClient(
bot_token=settings.TELEGRAM_BOT_TOKEN,
chat_id=settings.TELEGRAM_CHAT_ID,
enabled=settings.TELEGRAM_ENABLED,
)
# Initialize volatility hunter
volatility_analyzer = VolatilityAnalyzer(min_volume_surge=2.0, min_price_change=1.0)
market_scanner = MarketScanner(
@@ -289,6 +332,9 @@ async def run(settings: Settings) -> None:
# Track last scan time for each market
last_scan_time: dict[str, float] = {}
# Track market open/close state for notifications
_market_states: dict[str, bool] = {} # market_code -> is_open
shutdown = asyncio.Event()
def _signal_handler() -> None:
@@ -302,12 +348,31 @@ async def run(settings: Settings) -> None:
logger.info("The Ouroboros is alive. Mode: %s", settings.MODE)
logger.info("Enabled markets: %s", settings.enabled_market_list)
# Notify system startup
try:
await telegram.notify_system_start(settings.MODE, settings.enabled_market_list)
except Exception as exc:
logger.warning("System startup notification failed: %s", exc)
try:
while not shutdown.is_set():
# Get currently open markets
open_markets = get_open_markets(settings.enabled_market_list)
if not open_markets:
# Notify market close for any markets that were open
for market_code, is_open in list(_market_states.items()):
if is_open:
try:
from src.markets.schedule import MARKETS
market_info = MARKETS.get(market_code)
if market_info:
await telegram.notify_market_close(market_info.name, 0.0)
except Exception as exc:
logger.warning("Market close notification failed: %s", exc)
_market_states[market_code] = False
# No markets open — wait until next market opens
try:
next_market, next_open_time = get_next_market_open(
@@ -333,6 +398,14 @@ async def run(settings: Settings) -> None:
if shutdown.is_set():
break
# Notify market open if it just opened
if not _market_states.get(market.code, False):
try:
await telegram.notify_market_open(market.name)
except Exception as exc:
logger.warning("Market open notification failed: %s", exc)
_market_states[market.code] = True
# Volatility Hunter: Scan market periodically to update watchlist
now_timestamp = asyncio.get_event_loop().time()
last_scan = last_scan_time.get(market.code, 0.0)
@@ -391,12 +464,22 @@ async def run(settings: Settings) -> None:
decision_logger,
context_store,
criticality_assessor,
telegram,
market,
stock_code,
)
break # Success — exit retry loop
except CircuitBreakerTripped:
except CircuitBreakerTripped as exc:
logger.critical("Circuit breaker tripped — shutting down")
try:
await telegram.notify_circuit_breaker(
pnl_pct=exc.pnl_pct,
threshold=exc.threshold,
)
except Exception as notify_exc:
logger.warning(
"Circuit breaker notification failed: %s", notify_exc
)
raise
except ConnectionError as exc:
if attempt < MAX_CONNECTION_RETRIES:

View File

@@ -49,6 +49,46 @@ class TestTokenManagement:
await broker.close()
@pytest.mark.asyncio
async def test_concurrent_token_refresh_calls_api_once(self, settings):
"""Multiple concurrent token requests should only call API once."""
broker = KISBroker(settings)
# Track how many times the mock API is called
call_count = [0]
def create_mock_resp():
call_count[0] += 1
mock_resp = AsyncMock()
mock_resp.status = 200
mock_resp.json = AsyncMock(
return_value={
"access_token": "tok_concurrent",
"token_type": "Bearer",
"expires_in": 86400,
}
)
mock_resp.__aenter__ = AsyncMock(return_value=mock_resp)
mock_resp.__aexit__ = AsyncMock(return_value=False)
return mock_resp
with patch("aiohttp.ClientSession.post", return_value=create_mock_resp()):
# Launch 5 concurrent token requests
tokens = await asyncio.gather(
broker._ensure_token(),
broker._ensure_token(),
broker._ensure_token(),
broker._ensure_token(),
broker._ensure_token(),
)
# All should get the same token
assert all(t == "tok_concurrent" for t in tokens)
# API should be called only once (due to lock)
assert call_count[0] == 1
await broker.close()
# ---------------------------------------------------------------------------
# Network Error Handling

561
tests/test_main.py Normal file
View File

@@ -0,0 +1,561 @@
"""Tests for main trading loop telegram integration."""
import asyncio
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from src.core.risk_manager import CircuitBreakerTripped, FatFingerRejected
from src.main import trading_cycle
class TestTradingCycleTelegramIntegration:
"""Test telegram notifications in trading_cycle function."""
@pytest.fixture
def mock_broker(self) -> MagicMock:
"""Create mock broker."""
broker = MagicMock()
broker.get_orderbook = AsyncMock(
return_value={
"output1": {
"stck_prpr": "50000",
"frgn_ntby_qty": "100",
}
}
)
broker.get_balance = AsyncMock(
return_value={
"output2": [
{
"tot_evlu_amt": "10000000",
"dnca_tot_amt": "5000000",
"pchs_amt_smtl_amt": "5000000",
}
]
}
)
broker.send_order = AsyncMock(return_value={"msg1": "OK"})
return broker
@pytest.fixture
def mock_overseas_broker(self) -> MagicMock:
"""Create mock overseas broker."""
broker = MagicMock()
return broker
@pytest.fixture
def mock_brain(self) -> MagicMock:
"""Create mock brain that decides to buy."""
brain = MagicMock()
decision = MagicMock()
decision.action = "BUY"
decision.confidence = 85
decision.rationale = "Test buy"
brain.decide = AsyncMock(return_value=decision)
return brain
@pytest.fixture
def mock_risk(self) -> MagicMock:
"""Create mock risk manager."""
risk = MagicMock()
risk.validate_order = MagicMock()
return risk
@pytest.fixture
def mock_db(self) -> MagicMock:
"""Create mock database connection."""
return MagicMock()
@pytest.fixture
def mock_decision_logger(self) -> MagicMock:
"""Create mock decision logger."""
logger = MagicMock()
logger.log_decision = MagicMock()
return logger
@pytest.fixture
def mock_context_store(self) -> MagicMock:
"""Create mock context store."""
store = MagicMock()
store.get_latest_timeframe = MagicMock(return_value=None)
return store
@pytest.fixture
def mock_criticality_assessor(self) -> MagicMock:
"""Create mock criticality assessor."""
assessor = MagicMock()
assessor.assess_market_conditions = MagicMock(
return_value=MagicMock(value="NORMAL")
)
assessor.get_timeout = MagicMock(return_value=5.0)
return assessor
@pytest.fixture
def mock_telegram(self) -> MagicMock:
"""Create mock telegram client."""
telegram = MagicMock()
telegram.notify_trade_execution = AsyncMock()
telegram.notify_fat_finger = AsyncMock()
telegram.notify_circuit_breaker = AsyncMock()
return telegram
@pytest.fixture
def mock_market(self) -> MagicMock:
"""Create mock market info."""
market = MagicMock()
market.name = "Korea"
market.code = "KR"
market.exchange_code = "KRX"
market.is_domestic = True
return market
@pytest.mark.asyncio
async def test_trade_execution_notification_sent(
self,
mock_broker: MagicMock,
mock_overseas_broker: MagicMock,
mock_brain: MagicMock,
mock_risk: MagicMock,
mock_db: MagicMock,
mock_decision_logger: MagicMock,
mock_context_store: MagicMock,
mock_criticality_assessor: MagicMock,
mock_telegram: MagicMock,
mock_market: MagicMock,
) -> None:
"""Test telegram notification sent on trade execution."""
with patch("src.main.log_trade"):
await trading_cycle(
broker=mock_broker,
overseas_broker=mock_overseas_broker,
brain=mock_brain,
risk=mock_risk,
db_conn=mock_db,
decision_logger=mock_decision_logger,
context_store=mock_context_store,
criticality_assessor=mock_criticality_assessor,
telegram=mock_telegram,
market=mock_market,
stock_code="005930",
)
# Verify notification was sent
mock_telegram.notify_trade_execution.assert_called_once()
call_kwargs = mock_telegram.notify_trade_execution.call_args.kwargs
assert call_kwargs["stock_code"] == "005930"
assert call_kwargs["market"] == "Korea"
assert call_kwargs["action"] == "BUY"
assert call_kwargs["confidence"] == 85
@pytest.mark.asyncio
async def test_trade_execution_notification_failure_doesnt_crash(
self,
mock_broker: MagicMock,
mock_overseas_broker: MagicMock,
mock_brain: MagicMock,
mock_risk: MagicMock,
mock_db: MagicMock,
mock_decision_logger: MagicMock,
mock_context_store: MagicMock,
mock_criticality_assessor: MagicMock,
mock_telegram: MagicMock,
mock_market: MagicMock,
) -> None:
"""Test trading continues even if notification fails."""
# Make notification fail
mock_telegram.notify_trade_execution.side_effect = Exception("API error")
with patch("src.main.log_trade"):
# Should not raise exception
await trading_cycle(
broker=mock_broker,
overseas_broker=mock_overseas_broker,
brain=mock_brain,
risk=mock_risk,
db_conn=mock_db,
decision_logger=mock_decision_logger,
context_store=mock_context_store,
criticality_assessor=mock_criticality_assessor,
telegram=mock_telegram,
market=mock_market,
stock_code="005930",
)
# Verify notification was attempted
mock_telegram.notify_trade_execution.assert_called_once()
@pytest.mark.asyncio
async def test_fat_finger_notification_sent(
self,
mock_broker: MagicMock,
mock_overseas_broker: MagicMock,
mock_brain: MagicMock,
mock_risk: MagicMock,
mock_db: MagicMock,
mock_decision_logger: MagicMock,
mock_context_store: MagicMock,
mock_criticality_assessor: MagicMock,
mock_telegram: MagicMock,
mock_market: MagicMock,
) -> None:
"""Test telegram notification sent on fat finger rejection."""
# Make risk manager reject the order
mock_risk.validate_order.side_effect = FatFingerRejected(
order_amount=2000000,
total_cash=5000000,
max_pct=30.0,
)
with patch("src.main.log_trade"):
with pytest.raises(FatFingerRejected):
await trading_cycle(
broker=mock_broker,
overseas_broker=mock_overseas_broker,
brain=mock_brain,
risk=mock_risk,
db_conn=mock_db,
decision_logger=mock_decision_logger,
context_store=mock_context_store,
criticality_assessor=mock_criticality_assessor,
telegram=mock_telegram,
market=mock_market,
stock_code="005930",
)
# Verify notification was sent
mock_telegram.notify_fat_finger.assert_called_once()
call_kwargs = mock_telegram.notify_fat_finger.call_args.kwargs
assert call_kwargs["stock_code"] == "005930"
assert call_kwargs["order_amount"] == 2000000
assert call_kwargs["total_cash"] == 5000000
assert call_kwargs["max_pct"] == 30.0
@pytest.mark.asyncio
async def test_fat_finger_notification_failure_still_raises(
self,
mock_broker: MagicMock,
mock_overseas_broker: MagicMock,
mock_brain: MagicMock,
mock_risk: MagicMock,
mock_db: MagicMock,
mock_decision_logger: MagicMock,
mock_context_store: MagicMock,
mock_criticality_assessor: MagicMock,
mock_telegram: MagicMock,
mock_market: MagicMock,
) -> None:
"""Test fat finger exception still raised even if notification fails."""
# Make risk manager reject the order
mock_risk.validate_order.side_effect = FatFingerRejected(
order_amount=2000000,
total_cash=5000000,
max_pct=30.0,
)
# Make notification fail
mock_telegram.notify_fat_finger.side_effect = Exception("API error")
with patch("src.main.log_trade"):
with pytest.raises(FatFingerRejected):
await trading_cycle(
broker=mock_broker,
overseas_broker=mock_overseas_broker,
brain=mock_brain,
risk=mock_risk,
db_conn=mock_db,
decision_logger=mock_decision_logger,
context_store=mock_context_store,
criticality_assessor=mock_criticality_assessor,
telegram=mock_telegram,
market=mock_market,
stock_code="005930",
)
# Verify notification was attempted
mock_telegram.notify_fat_finger.assert_called_once()
@pytest.mark.asyncio
async def test_no_notification_on_hold_decision(
self,
mock_broker: MagicMock,
mock_overseas_broker: MagicMock,
mock_brain: MagicMock,
mock_risk: MagicMock,
mock_db: MagicMock,
mock_decision_logger: MagicMock,
mock_context_store: MagicMock,
mock_criticality_assessor: MagicMock,
mock_telegram: MagicMock,
mock_market: MagicMock,
) -> None:
"""Test no trade notification sent when decision is HOLD."""
# Change brain decision to HOLD
decision = MagicMock()
decision.action = "HOLD"
decision.confidence = 50
decision.rationale = "Insufficient signal"
mock_brain.decide = AsyncMock(return_value=decision)
with patch("src.main.log_trade"):
await trading_cycle(
broker=mock_broker,
overseas_broker=mock_overseas_broker,
brain=mock_brain,
risk=mock_risk,
db_conn=mock_db,
decision_logger=mock_decision_logger,
context_store=mock_context_store,
criticality_assessor=mock_criticality_assessor,
telegram=mock_telegram,
market=mock_market,
stock_code="005930",
)
# Verify no trade notification sent
mock_telegram.notify_trade_execution.assert_not_called()
class TestRunFunctionTelegramIntegration:
"""Test telegram notifications in run function."""
@pytest.mark.asyncio
async def test_circuit_breaker_notification_sent(self) -> None:
"""Test telegram notification sent when circuit breaker trips."""
mock_telegram = MagicMock()
mock_telegram.notify_circuit_breaker = AsyncMock()
# Simulate circuit breaker exception
exc = CircuitBreakerTripped(pnl_pct=-3.5, threshold=-3.0)
# Test the notification logic
try:
await mock_telegram.notify_circuit_breaker(
pnl_pct=exc.pnl_pct,
threshold=exc.threshold,
)
except Exception:
pass # Ignore errors in notification
# Verify notification was called
mock_telegram.notify_circuit_breaker.assert_called_once_with(
pnl_pct=-3.5,
threshold=-3.0,
)
class TestOverseasBalanceParsing:
"""Test overseas balance output2 parsing handles different formats."""
@pytest.fixture
def mock_overseas_broker_with_list(self) -> MagicMock:
"""Create mock overseas broker returning list format."""
broker = MagicMock()
broker.get_overseas_price = AsyncMock(
return_value={"output": {"last": "150.50"}}
)
broker.get_overseas_balance = AsyncMock(
return_value={
"output2": [
{
"frcr_evlu_tota": "10000.00",
"frcr_dncl_amt_2": "5000.00",
"frcr_buy_amt_smtl": "4500.00",
}
]
}
)
return broker
@pytest.fixture
def mock_overseas_broker_with_dict(self) -> MagicMock:
"""Create mock overseas broker returning dict format."""
broker = MagicMock()
broker.get_overseas_price = AsyncMock(
return_value={"output": {"last": "150.50"}}
)
broker.get_overseas_balance = AsyncMock(
return_value={
"output2": {
"frcr_evlu_tota": "10000.00",
"frcr_dncl_amt_2": "5000.00",
"frcr_buy_amt_smtl": "4500.00",
}
}
)
return broker
@pytest.fixture
def mock_overseas_broker_with_empty(self) -> MagicMock:
"""Create mock overseas broker returning empty output2."""
broker = MagicMock()
broker.get_overseas_price = AsyncMock(
return_value={"output": {"last": "150.50"}}
)
broker.get_overseas_balance = AsyncMock(return_value={"output2": []})
return broker
@pytest.fixture
def mock_domestic_broker(self) -> MagicMock:
"""Create minimal mock domestic broker."""
broker = MagicMock()
return broker
@pytest.fixture
def mock_overseas_market(self) -> MagicMock:
"""Create mock overseas market info."""
market = MagicMock()
market.name = "NASDAQ"
market.code = "US_NASDAQ"
market.exchange_code = "NASD"
market.is_domestic = False
return market
@pytest.fixture
def mock_brain_hold(self) -> MagicMock:
"""Create mock brain that always holds."""
brain = MagicMock()
decision = MagicMock()
decision.action = "HOLD"
decision.confidence = 50
decision.rationale = "Testing balance parsing"
brain.decide = AsyncMock(return_value=decision)
return brain
@pytest.fixture
def mock_risk(self) -> MagicMock:
"""Create mock risk manager."""
return MagicMock()
@pytest.fixture
def mock_db(self) -> MagicMock:
"""Create mock database."""
return MagicMock()
@pytest.fixture
def mock_decision_logger(self) -> MagicMock:
"""Create mock decision logger."""
return MagicMock()
@pytest.fixture
def mock_context_store(self) -> MagicMock:
"""Create mock context store."""
store = MagicMock()
store.get_latest_timeframe = MagicMock(return_value=None)
return store
@pytest.fixture
def mock_criticality_assessor(self) -> MagicMock:
"""Create mock criticality assessor."""
assessor = MagicMock()
assessor.assess_market_conditions = MagicMock(
return_value=MagicMock(value="NORMAL")
)
assessor.get_timeout = MagicMock(return_value=5.0)
return assessor
@pytest.fixture
def mock_telegram(self) -> MagicMock:
"""Create mock telegram client."""
return MagicMock()
@pytest.mark.asyncio
async def test_overseas_balance_list_format(
self,
mock_domestic_broker: MagicMock,
mock_overseas_broker_with_list: MagicMock,
mock_brain_hold: MagicMock,
mock_risk: MagicMock,
mock_db: MagicMock,
mock_decision_logger: MagicMock,
mock_context_store: MagicMock,
mock_criticality_assessor: MagicMock,
mock_telegram: MagicMock,
mock_overseas_market: MagicMock,
) -> None:
"""Test overseas balance parsing with list format (output2=[{...}])."""
with patch("src.main.log_trade"):
# Should not raise KeyError
await trading_cycle(
broker=mock_domestic_broker,
overseas_broker=mock_overseas_broker_with_list,
brain=mock_brain_hold,
risk=mock_risk,
db_conn=mock_db,
decision_logger=mock_decision_logger,
context_store=mock_context_store,
criticality_assessor=mock_criticality_assessor,
telegram=mock_telegram,
market=mock_overseas_market,
stock_code="AAPL",
)
# Verify balance API was called
mock_overseas_broker_with_list.get_overseas_balance.assert_called_once()
@pytest.mark.asyncio
async def test_overseas_balance_dict_format(
self,
mock_domestic_broker: MagicMock,
mock_overseas_broker_with_dict: MagicMock,
mock_brain_hold: MagicMock,
mock_risk: MagicMock,
mock_db: MagicMock,
mock_decision_logger: MagicMock,
mock_context_store: MagicMock,
mock_criticality_assessor: MagicMock,
mock_telegram: MagicMock,
mock_overseas_market: MagicMock,
) -> None:
"""Test overseas balance parsing with dict format (output2={...})."""
with patch("src.main.log_trade"):
# Should not raise KeyError
await trading_cycle(
broker=mock_domestic_broker,
overseas_broker=mock_overseas_broker_with_dict,
brain=mock_brain_hold,
risk=mock_risk,
db_conn=mock_db,
decision_logger=mock_decision_logger,
context_store=mock_context_store,
criticality_assessor=mock_criticality_assessor,
telegram=mock_telegram,
market=mock_overseas_market,
stock_code="AAPL",
)
# Verify balance API was called
mock_overseas_broker_with_dict.get_overseas_balance.assert_called_once()
@pytest.mark.asyncio
async def test_overseas_balance_empty_format(
self,
mock_domestic_broker: MagicMock,
mock_overseas_broker_with_empty: MagicMock,
mock_brain_hold: MagicMock,
mock_risk: MagicMock,
mock_db: MagicMock,
mock_decision_logger: MagicMock,
mock_context_store: MagicMock,
mock_criticality_assessor: MagicMock,
mock_telegram: MagicMock,
mock_overseas_market: MagicMock,
) -> None:
"""Test overseas balance parsing with empty output2."""
with patch("src.main.log_trade"):
# Should not raise KeyError, should default to 0
await trading_cycle(
broker=mock_domestic_broker,
overseas_broker=mock_overseas_broker_with_empty,
brain=mock_brain_hold,
risk=mock_risk,
db_conn=mock_db,
decision_logger=mock_decision_logger,
context_store=mock_context_store,
criticality_assessor=mock_criticality_assessor,
telegram=mock_telegram,
market=mock_overseas_market,
stock_code="AAPL",
)
# Verify balance API was called
mock_overseas_broker_with_empty.get_overseas_balance.assert_called_once()