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 # Database
DB_PATH=data/trade_logs.db DB_PATH=data/trade_logs.db
# Rate Limiting # Rate Limiting (requests per second for KIS API)
RATE_LIMIT_RPS=10.0 # Reduced to 5.0 to avoid "초당 거래건수 초과" errors (EGW00201)
RATE_LIMIT_RPS=5.0
# Trading Mode (paper / live) # Trading Mode (paper / live)
MODE=paper MODE=paper
@@ -26,3 +27,10 @@ MODE=paper
# NEWS_API_KEY=your_news_api_key_here # NEWS_API_KEY=your_news_api_key_here
# NEWS_API_PROVIDER=alphavantage # NEWS_API_PROVIDER=alphavantage
# MARKET_DATA_API_KEY=your_market_data_key_here # 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 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 ## Documentation
- **[Workflow Guide](docs/workflow.md)** — Git workflow policy and agent-based development - **[Workflow Guide](docs/workflow.md)** — Git workflow policy and agent-based development
@@ -42,11 +70,12 @@ src/
├── core/ # Risk manager (READ-ONLY) ├── core/ # Risk manager (READ-ONLY)
├── evolution/ # Self-improvement optimizer ├── evolution/ # Self-improvement optimizer
├── markets/ # Market schedules and timezone handling ├── markets/ # Market schedules and timezone handling
├── notifications/ # Telegram real-time alerts
├── db.py # SQLite trade logging ├── db.py # SQLite trade logging
├── main.py # Trading loop orchestrator ├── main.py # Trading loop orchestrator
└── config.py # Settings (from .env) └── config.py # Settings (from .env)
tests/ # 54 tests across 4 files tests/ # 273 tests across 13 files
docs/ # Extended documentation docs/ # Extended documentation
``` ```

View File

@@ -29,6 +29,7 @@ KIS(한국투자증권) API로 매매하고, Google Gemini로 판단하며, 자
| 브로커 | `src/broker/kis_api.py` | KIS API 비동기 래퍼 (토큰 갱신, 레이트 리미터, 해시키) | | 브로커 | `src/broker/kis_api.py` | KIS API 비동기 래퍼 (토큰 갱신, 레이트 리미터, 해시키) |
| 두뇌 | `src/brain/gemini_client.py` | Gemini 프롬프트 구성 및 JSON 응답 파싱 | | 두뇌 | `src/brain/gemini_client.py` | Gemini 프롬프트 구성 및 JSON 응답 파싱 |
| 방패 | `src/core/risk_manager.py` | 서킷 브레이커 + 팻 핑거 체크 | | 방패 | `src/core/risk_manager.py` | 서킷 브레이커 + 팻 핑거 체크 |
| 알림 | `src/notifications/telegram_client.py` | 텔레그램 실시간 거래 알림 (선택사항) |
| 진화 | `src/evolution/optimizer.py` | 실패 패턴 분석 → 새 전략 생성 → 테스트 → PR | | 진화 | `src/evolution/optimizer.py` | 실패 패턴 분석 → 새 전략 생성 → 테스트 → PR |
| DB | `src/db.py` | SQLite 거래 로그 기록 | | DB | `src/db.py` | SQLite 거래 로그 기록 |
@@ -75,6 +76,34 @@ python -m src.main --mode=paper
docker compose up -d ouroboros 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 방식으로 구현 전에 먼저 작성되었습니다. 35개 테스트가 TDD 방식으로 구현 전에 먼저 작성되었습니다.
@@ -111,6 +140,7 @@ The-Ouroboros/
│ ├── broker/kis_api.py # KIS API 클라이언트 │ ├── broker/kis_api.py # KIS API 클라이언트
│ ├── brain/gemini_client.py # Gemini 의사결정 엔진 │ ├── brain/gemini_client.py # Gemini 의사결정 엔진
│ ├── core/risk_manager.py # 리스크 관리 │ ├── core/risk_manager.py # 리스크 관리
│ ├── notifications/telegram_client.py # 텔레그램 알림
│ ├── evolution/optimizer.py # 전략 진화 엔진 │ ├── evolution/optimizer.py # 전략 진화 엔진
│ └── strategies/base.py # 전략 베이스 클래스 │ └── strategies/base.py # 전략 베이스 클래스
├── tests/ # TDD 테스트 스위트 ├── tests/ # TDD 테스트 스위트

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 - **Fat-Finger Protection**: Rejects orders exceeding 30% of available cash
- Must always be enforced, cannot be disabled - 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 **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 │ │ Database: Log Trade │
│ - SQLite (data/trades.db) │ │ - SQLite (data/trades.db) │
│ - Track: action, confidence, │ │ - Track: action, confidence, │
@@ -164,6 +191,11 @@ CONFIDENCE_THRESHOLD=80
MAX_LOSS_PCT=3.0 MAX_LOSS_PCT=3.0
MAX_ORDER_PCT=30.0 MAX_ORDER_PCT=30.0
ENABLED_MARKETS=KR,US_NASDAQ # Comma-separated market codes 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`. 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 - Wait until next market opens
- Use `get_next_market_open()` to calculate wait time - Use `get_next_market_open()` to calculate wait time
- Sleep until market open 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._session: aiohttp.ClientSession | None = None
self._access_token: str | None = None self._access_token: str | None = None
self._token_expires_at: float = 0.0 self._token_expires_at: float = 0.0
self._token_lock = asyncio.Lock()
self._rate_limiter = LeakyBucket(settings.RATE_LIMIT_RPS) self._rate_limiter = LeakyBucket(settings.RATE_LIMIT_RPS)
def _get_session(self) -> aiohttp.ClientSession: def _get_session(self) -> aiohttp.ClientSession:
@@ -80,7 +81,19 @@ class KISBroker:
# ------------------------------------------------------------------ # ------------------------------------------------------------------
async def _ensure_token(self) -> str: 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
# 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() now = asyncio.get_event_loop().time()
if self._access_token and now < self._token_expires_at: if self._access_token and now < self._token_expires_at:
return self._access_token return self._access_token

View File

@@ -37,7 +37,8 @@ class Settings(BaseSettings):
DB_PATH: str = "data/trade_logs.db" DB_PATH: str = "data/trade_logs.db"
# Rate Limiting (requests per second for KIS API) # 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 # Trading mode
MODE: str = Field(default="paper", pattern="^(paper|live)$") MODE: str = Field(default="paper", pattern="^(paper|live)$")
@@ -54,6 +55,11 @@ class Settings(BaseSettings):
S3_BUCKET_NAME: str | None = None S3_BUCKET_NAME: str | None = None
S3_REGION: str = "us-east-1" 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"} model_config = {"env_file": ".env", "env_file_encoding": "utf-8"}
@property @property

View File

@@ -10,6 +10,7 @@ import argparse
import asyncio import asyncio
import logging import logging
import signal import signal
import sys
from datetime import UTC, datetime from datetime import UTC, datetime
from typing import Any from typing import Any
@@ -23,11 +24,12 @@ from src.context.layer import ContextLayer
from src.context.store import ContextStore from src.context.store import ContextStore
from src.core.criticality import CriticalityAssessor from src.core.criticality import CriticalityAssessor
from src.core.priority_queue import PriorityTaskQueue 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.db import init_db, log_trade
from src.logging.decision_logger import DecisionLogger from src.logging.decision_logger import DecisionLogger
from src.logging_config import setup_logging from src.logging_config import setup_logging
from src.markets.schedule import MarketInfo, get_next_market_open, get_open_markets from src.markets.schedule import MarketInfo, get_next_market_open, get_open_markets
from src.notifications.telegram_client import TelegramClient
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -62,6 +64,7 @@ async def trading_cycle(
decision_logger: DecisionLogger, decision_logger: DecisionLogger,
context_store: ContextStore, context_store: ContextStore,
criticality_assessor: CriticalityAssessor, criticality_assessor: CriticalityAssessor,
telegram: TelegramClient,
market: MarketInfo, market: MarketInfo,
stock_code: str, stock_code: str,
) -> None: ) -> None:
@@ -92,9 +95,17 @@ async def trading_cycle(
balance_data = await overseas_broker.get_overseas_balance(market.exchange_code) balance_data = await overseas_broker.get_overseas_balance(market.exchange_code)
output2 = balance_data.get("output2", [{}]) output2 = balance_data.get("output2", [{}])
total_eval = float(output2[0].get("frcr_evlu_tota", "0")) if output2 else 0 # Handle both list and dict response formats
total_cash = float(output2[0].get("frcr_dncl_amt_2", "0")) if output2 else 0 if isinstance(output2, list) and output2:
purchase_total = float(output2[0].get("frcr_buy_amt_smtl", "0")) if output2 else 0 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")) current_price = float(price_data.get("output", {}).get("last", "0"))
foreigner_net = 0.0 # Not available for overseas foreigner_net = 0.0 # Not available for overseas
@@ -199,11 +210,23 @@ async def trading_cycle(
order_amount = current_price * quantity order_amount = current_price * quantity
# 4. Risk check BEFORE order # 4. Risk check BEFORE order
try:
risk.validate_order( risk.validate_order(
current_pnl_pct=pnl_pct, current_pnl_pct=pnl_pct,
order_amount=order_amount, order_amount=order_amount,
total_cash=total_cash, 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 # 5. Send order
if market.is_domestic: if market.is_domestic:
@@ -223,6 +246,19 @@ async def trading_cycle(
) )
logger.info("Order result: %s", result.get("msg1", "OK")) 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 # 6. Log trade
log_trade( log_trade(
conn=db_conn, conn=db_conn,
@@ -266,6 +302,13 @@ async def run(settings: Settings) -> None:
decision_logger = DecisionLogger(db_conn) decision_logger = DecisionLogger(db_conn)
context_store = ContextStore(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 # Initialize volatility hunter
volatility_analyzer = VolatilityAnalyzer(min_volume_surge=2.0, min_price_change=1.0) volatility_analyzer = VolatilityAnalyzer(min_volume_surge=2.0, min_price_change=1.0)
market_scanner = MarketScanner( market_scanner = MarketScanner(
@@ -289,6 +332,9 @@ async def run(settings: Settings) -> None:
# Track last scan time for each market # Track last scan time for each market
last_scan_time: dict[str, float] = {} 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() shutdown = asyncio.Event()
def _signal_handler() -> None: 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("The Ouroboros is alive. Mode: %s", settings.MODE)
logger.info("Enabled markets: %s", settings.enabled_market_list) 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: try:
while not shutdown.is_set(): while not shutdown.is_set():
# Get currently open markets # Get currently open markets
open_markets = get_open_markets(settings.enabled_market_list) open_markets = get_open_markets(settings.enabled_market_list)
if not open_markets: 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 # No markets open — wait until next market opens
try: try:
next_market, next_open_time = get_next_market_open( next_market, next_open_time = get_next_market_open(
@@ -333,6 +398,14 @@ async def run(settings: Settings) -> None:
if shutdown.is_set(): if shutdown.is_set():
break 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 # Volatility Hunter: Scan market periodically to update watchlist
now_timestamp = asyncio.get_event_loop().time() now_timestamp = asyncio.get_event_loop().time()
last_scan = last_scan_time.get(market.code, 0.0) last_scan = last_scan_time.get(market.code, 0.0)
@@ -391,12 +464,22 @@ async def run(settings: Settings) -> None:
decision_logger, decision_logger,
context_store, context_store,
criticality_assessor, criticality_assessor,
telegram,
market, market,
stock_code, stock_code,
) )
break # Success — exit retry loop break # Success — exit retry loop
except CircuitBreakerTripped: except CircuitBreakerTripped as exc:
logger.critical("Circuit breaker tripped — shutting down") 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 raise
except ConnectionError as exc: except ConnectionError as exc:
if attempt < MAX_CONNECTION_RETRIES: if attempt < MAX_CONNECTION_RETRIES:

View File

@@ -49,6 +49,46 @@ class TestTokenManagement:
await broker.close() 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 # 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()