diff --git a/CLAUDE.md b/CLAUDE.md index f081363..7c3a8c1 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -53,6 +53,7 @@ Get real-time alerts for trades, circuit breakers, and system events via Telegra - **[Context Tree](docs/context-tree.md)** — L1-L7 hierarchical memory system - **[Testing](docs/testing.md)** — Test structure, coverage requirements, writing tests - **[Agent Policies](docs/agents.md)** — Prime directives, constraints, prohibited actions +- **[Requirements Log](docs/requirements-log.md)** — User requirements and feedback tracking ## Core Principles @@ -61,6 +62,15 @@ Get real-time alerts for trades, circuit breakers, and system events via Telegra 3. **Issue-Driven Development** — All work goes through Gitea issues → feature branches → PRs 4. **Agent Specialization** — Use dedicated agents for design, coding, testing, docs, review +## Requirements Management + +User requirements and feedback are tracked in [docs/requirements-log.md](docs/requirements-log.md): + +- New requirements are added chronologically with dates +- Code changes should reference related requirements +- Helps maintain project evolution aligned with user needs +- Preserves context across conversations and development cycles + ## Project Structure ``` diff --git a/docs/architecture.md b/docs/architecture.md index 06bf31a..a8e23d4 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -2,7 +2,42 @@ ## Overview -Self-evolving AI trading agent for global stock markets via KIS (Korea Investment & Securities) API. The main loop in `src/main.py` orchestrates four components in a 60-second cycle per stock across multiple markets. +Self-evolving AI trading agent for global stock markets via KIS (Korea Investment & Securities) API. The main loop in `src/main.py` orchestrates four components across multiple markets with two trading modes: daily (batch API calls) or realtime (per-stock decisions). + +## Trading Modes + +The system supports two trading frequency modes controlled by the `TRADE_MODE` environment variable: + +### Daily Mode (default) + +Optimized for Gemini Free tier API limits (20 calls/day): + +- **Batch decisions**: 1 API call per market per session +- **Fixed schedule**: 4 sessions per day at 6-hour intervals (configurable) +- **API efficiency**: Processes all stocks in a market simultaneously +- **Use case**: Free tier users, cost-conscious deployments +- **Configuration**: + ```bash + TRADE_MODE=daily + DAILY_SESSIONS=4 # Sessions per day (1-10) + SESSION_INTERVAL_HOURS=6 # Hours between sessions (1-24) + ``` + +**Example**: With 2 markets (US, KR) and 4 sessions/day = 8 API calls/day (within 20 call limit) + +### Realtime Mode + +High-frequency trading with individual stock analysis: + +- **Per-stock decisions**: 1 API call per stock per cycle +- **60-second interval**: Continuous monitoring +- **Use case**: Production deployments with Gemini paid tier +- **Configuration**: + ```bash + TRADE_MODE=realtime + ``` + +**Note**: Realtime mode requires Gemini API subscription due to high call volume. ## Core Components @@ -192,6 +227,11 @@ MAX_LOSS_PCT=3.0 MAX_ORDER_PCT=30.0 ENABLED_MARKETS=KR,US_NASDAQ # Comma-separated market codes +# Trading Mode (API efficiency) +TRADE_MODE=daily # daily | realtime +DAILY_SESSIONS=4 # Sessions per day (daily mode only) +SESSION_INTERVAL_HOURS=6 # Hours between sessions (daily mode only) + # Telegram Notifications (optional) TELEGRAM_BOT_TOKEN=1234567890:ABCdefGHIjklMNOpqrsTUVwxyz TELEGRAM_CHAT_ID=123456789 diff --git a/docs/requirements-log.md b/docs/requirements-log.md new file mode 100644 index 0000000..2fbcfd8 --- /dev/null +++ b/docs/requirements-log.md @@ -0,0 +1,28 @@ +# Requirements Log + +프로젝트 진화를 위한 사용자 요구사항 기록. + +이 문서는 시간순으로 사용자와의 대화에서 나온 요구사항과 피드백을 기록합니다. +새로운 요구사항이 있으면 날짜와 함께 추가하세요. + +--- + +## 2026-02-05 + +### API 효율화 +- Gemini API는 귀중한 자원. 종목별 개별 호출 대신 배치 호출 필요 +- Free tier 한도(20 calls/day) 고려하여 일일 몇 차례 거래 모드로 전환 +- 배치 API 호출로 여러 종목을 한 번에 분석 + +### 거래 모드 +- **Daily Mode**: 하루 4회 거래 세션 (6시간 간격) - Free tier 호환 +- **Realtime Mode**: 60초 간격 실시간 거래 - 유료 구독 필요 +- `TRADE_MODE` 환경변수로 모드 선택 + +### 진화 시스템 +- 사용자 대화 내용을 문서로 기록하여 향후에도 의도 반영 +- 프롬프트 품질 검증은 별도 이슈로 다룰 예정 + +### 문서화 +- 시스템 구조, 기능별 설명 등 코드 문서화 항상 신경쓸 것 +- 새로운 기능 추가 시 관련 문서 업데이트 필수 diff --git a/src/brain/gemini_client.py b/src/brain/gemini_client.py index 3c3db81..be41bbf 100644 --- a/src/brain/gemini_client.py +++ b/src/brain/gemini_client.py @@ -525,3 +525,233 @@ class GeminiClient: DecisionCache instance or None if caching disabled """ return self._cache + + # ------------------------------------------------------------------ + # Batch Decision Making (for daily trading mode) + # ------------------------------------------------------------------ + + async def decide_batch( + self, stocks_data: list[dict[str, Any]] + ) -> dict[str, TradeDecision]: + """Make decisions for multiple stocks in a single API call. + + This is designed for daily trading mode to minimize API usage + when working with Gemini Free tier (20 calls/day limit). + + Args: + stocks_data: List of market data dictionaries, each with: + - stock_code: Stock ticker + - current_price: Current price + - market_name: Market name (optional) + - foreigner_net: Foreigner net buy/sell (optional) + + Returns: + Dictionary mapping stock_code to TradeDecision + + Example: + >>> stocks_data = [ + ... {"stock_code": "AAPL", "current_price": 185.5}, + ... {"stock_code": "MSFT", "current_price": 420.0}, + ... ] + >>> decisions = await client.decide_batch(stocks_data) + >>> decisions["AAPL"].action + 'BUY' + """ + if not stocks_data: + return {} + + # Build compressed batch prompt + market_name = stocks_data[0].get("market_name", "stock market") + + # Format stock data as compact JSON array + compact_stocks = [] + for stock in stocks_data: + compact = { + "code": stock["stock_code"], + "price": stock["current_price"], + } + if stock.get("foreigner_net", 0) != 0: + compact["frgn"] = stock["foreigner_net"] + compact_stocks.append(compact) + + data_str = json.dumps(compact_stocks, ensure_ascii=False) + + prompt = ( + f"You are a professional {market_name} trading analyst.\n" + "Analyze the following stocks and decide whether to BUY, SELL, or HOLD each one.\n\n" + f"Stock Data: {data_str}\n\n" + "You MUST respond with ONLY a valid JSON array in this format:\n" + '[{"code": "AAPL", "action": "BUY", "confidence": 85, "rationale": "..."},\n' + ' {"code": "MSFT", "action": "HOLD", "confidence": 50, "rationale": "..."}, ...]\n\n' + "Rules:\n" + "- Return one decision object per stock\n" + "- action must be exactly: BUY, SELL, or HOLD\n" + "- confidence must be 0-100\n" + "- rationale should be concise (1-2 sentences)\n" + "- Do NOT wrap JSON in markdown code blocks\n" + ) + + # Estimate tokens + token_count = self._optimizer.estimate_tokens(prompt) + self._total_tokens_used += token_count + + logger.info( + "Requesting batch decision for %d stocks from Gemini", + len(stocks_data), + extra={"estimated_tokens": token_count}, + ) + + try: + response = await self._client.aio.models.generate_content( + model=self._model_name, + contents=prompt, + ) + raw = response.text + except Exception as exc: + logger.error("Gemini API error in batch decision: %s", exc) + # Return HOLD for all stocks on API error + return { + stock["stock_code"]: TradeDecision( + action="HOLD", + confidence=0, + rationale=f"API error: {exc}", + token_count=token_count, + cached=False, + ) + for stock in stocks_data + } + + # Parse batch response + return self._parse_batch_response(raw, stocks_data, token_count) + + def _parse_batch_response( + self, raw: str, stocks_data: list[dict[str, Any]], token_count: int + ) -> dict[str, TradeDecision]: + """Parse batch response into a dictionary of decisions. + + Args: + raw: Raw response from Gemini + stocks_data: Original stock data list + token_count: Token count for the request + + Returns: + Dictionary mapping stock_code to TradeDecision + """ + if not raw or not raw.strip(): + logger.warning("Empty batch response from Gemini — defaulting all to HOLD") + return { + stock["stock_code"]: TradeDecision( + action="HOLD", + confidence=0, + rationale="Empty response", + token_count=0, + cached=False, + ) + for stock in stocks_data + } + + # Strip markdown code fences if present + cleaned = raw.strip() + match = re.search(r"```(?:json)?\s*\n?(.*?)\n?```", cleaned, re.DOTALL) + if match: + cleaned = match.group(1).strip() + + try: + data = json.loads(cleaned) + except json.JSONDecodeError: + logger.warning("Malformed JSON in batch response — defaulting all to HOLD") + return { + stock["stock_code"]: TradeDecision( + action="HOLD", + confidence=0, + rationale="Malformed JSON response", + token_count=0, + cached=False, + ) + for stock in stocks_data + } + + if not isinstance(data, list): + logger.warning("Batch response is not a JSON array — defaulting all to HOLD") + return { + stock["stock_code"]: TradeDecision( + action="HOLD", + confidence=0, + rationale="Invalid response format", + token_count=0, + cached=False, + ) + for stock in stocks_data + } + + # Build decision map + decisions: dict[str, TradeDecision] = {} + stock_codes = {stock["stock_code"] for stock in stocks_data} + + for item in data: + if not isinstance(item, dict): + continue + + code = item.get("code") + if not code or code not in stock_codes: + continue + + # Validate required fields + if not all(k in item for k in ("action", "confidence", "rationale")): + logger.warning("Missing fields for %s — using HOLD", code) + decisions[code] = TradeDecision( + action="HOLD", + confidence=0, + rationale="Missing required fields", + token_count=0, + cached=False, + ) + continue + + action = str(item["action"]).upper() + if action not in VALID_ACTIONS: + logger.warning("Invalid action '%s' for %s — forcing HOLD", action, code) + action = "HOLD" + + confidence = int(item["confidence"]) + rationale = str(item["rationale"]) + + # Enforce confidence threshold + if confidence < self._confidence_threshold: + logger.info( + "Confidence %d < threshold %d for %s — forcing HOLD", + confidence, + self._confidence_threshold, + code, + ) + action = "HOLD" + + decisions[code] = TradeDecision( + action=action, + confidence=confidence, + rationale=rationale, + token_count=token_count // len(stocks_data), # Split token cost + cached=False, + ) + self._total_decisions += 1 + + # Fill in missing stocks with HOLD + for stock in stocks_data: + code = stock["stock_code"] + if code not in decisions: + logger.warning("No decision for %s in batch response — using HOLD", code) + decisions[code] = TradeDecision( + action="HOLD", + confidence=0, + rationale="Not found in batch response", + token_count=0, + cached=False, + ) + + logger.info( + "Batch decision completed for %d stocks", + len(decisions), + extra={"tokens": token_count}, + ) + + return decisions diff --git a/src/config.py b/src/config.py index 617270c..71ac372 100644 --- a/src/config.py +++ b/src/config.py @@ -44,6 +44,11 @@ class Settings(BaseSettings): # Trading mode MODE: str = Field(default="paper", pattern="^(paper|live)$") + # Trading frequency mode (daily = batch API calls, realtime = per-stock calls) + TRADE_MODE: str = Field(default="daily", pattern="^(daily|realtime)$") + DAILY_SESSIONS: int = Field(default=4, ge=1, le=10) + SESSION_INTERVAL_HOURS: int = Field(default=6, ge=1, le=24) + # Market selection (comma-separated market codes) ENABLED_MARKETS: str = "KR" diff --git a/src/main.py b/src/main.py index f4dbefd..58fb62d 100644 --- a/src/main.py +++ b/src/main.py @@ -74,6 +74,10 @@ TRADE_INTERVAL_SECONDS = 60 SCAN_INTERVAL_SECONDS = 60 # Scan markets every 60 seconds MAX_CONNECTION_RETRIES = 3 +# Daily trading mode constants (for Free tier API efficiency) +DAILY_TRADE_SESSIONS = 4 # Number of trading sessions per day +TRADE_SESSION_INTERVAL_HOURS = 6 # Hours between sessions + # Full stock universe per market (for scanning) # In production, this would be loaded from a database or API STOCK_UNIVERSE = { @@ -321,6 +325,239 @@ async def trading_cycle( ) +async def run_daily_session( + broker: KISBroker, + overseas_broker: OverseasBroker, + brain: GeminiClient, + risk: RiskManager, + db_conn: Any, + decision_logger: DecisionLogger, + context_store: ContextStore, + criticality_assessor: CriticalityAssessor, + telegram: TelegramClient, + settings: Settings, +) -> None: + """Execute one daily trading session. + + Designed for API efficiency with Gemini Free tier: + - Batch decision making (1 API call per market) + - Runs N times per day at fixed intervals + - Minimizes API usage while maintaining trading capability + """ + # Get currently open markets + open_markets = get_open_markets(settings.enabled_market_list) + + if not open_markets: + logger.info("No markets open for this session") + return + + logger.info("Starting daily trading session for %d markets", len(open_markets)) + + # Process each open market + for market in open_markets: + # Get watchlist for this market + watchlist = WATCHLISTS.get(market.code, []) + if not watchlist: + logger.debug("No watchlist for market %s", market.code) + continue + + logger.info("Processing market: %s (%d stocks)", market.name, len(watchlist)) + + # Collect market data for all stocks in the watchlist + stocks_data = [] + for stock_code in watchlist: + try: + if market.is_domestic: + orderbook = await broker.get_orderbook(stock_code) + current_price = safe_float(orderbook.get("output1", {}).get("stck_prpr", "0")) + foreigner_net = safe_float( + orderbook.get("output1", {}).get("frgn_ntby_qty", "0") + ) + else: + price_data = await overseas_broker.get_overseas_price( + market.exchange_code, stock_code + ) + current_price = safe_float(price_data.get("output", {}).get("last", "0")) + foreigner_net = 0.0 + + stocks_data.append( + { + "stock_code": stock_code, + "market_name": market.name, + "current_price": current_price, + "foreigner_net": foreigner_net, + } + ) + except Exception as exc: + logger.error("Failed to fetch data for %s: %s", stock_code, exc) + continue + + if not stocks_data: + logger.warning("No valid stock data for market %s", market.code) + continue + + # Get batch decisions (1 API call for all stocks in this market) + logger.info("Requesting batch decision for %d stocks in %s", len(stocks_data), market.name) + decisions = await brain.decide_batch(stocks_data) + + # Get balance data once for the market + if market.is_domestic: + balance_data = await broker.get_balance() + output2 = balance_data.get("output2", [{}]) + total_eval = safe_float(output2[0].get("tot_evlu_amt", "0")) if output2 else 0 + total_cash = safe_float(output2[0].get("dnca_tot_amt", "0")) if output2 else 0 + purchase_total = safe_float(output2[0].get("pchs_amt_smtl_amt", "0")) if output2 else 0 + else: + balance_data = await overseas_broker.get_overseas_balance(market.exchange_code) + output2 = balance_data.get("output2", [{}]) + if isinstance(output2, list) and output2: + balance_info = output2[0] + elif isinstance(output2, dict): + balance_info = output2 + else: + balance_info = {} + + total_eval = safe_float(balance_info.get("frcr_evlu_tota", "0") or "0") + total_cash = safe_float(balance_info.get("frcr_dncl_amt_2", "0") or "0") + purchase_total = safe_float(balance_info.get("frcr_buy_amt_smtl", "0") or "0") + + # Calculate daily P&L % + pnl_pct = ( + ((total_eval - purchase_total) / purchase_total * 100) if purchase_total > 0 else 0.0 + ) + + # Execute decisions for each stock + for stock_data in stocks_data: + stock_code = stock_data["stock_code"] + decision = decisions.get(stock_code) + + if not decision: + logger.warning("No decision for %s — skipping", stock_code) + continue + + logger.info( + "Decision for %s (%s): %s (confidence=%d)", + stock_code, + market.name, + decision.action, + decision.confidence, + ) + + # Log decision + context_snapshot = { + "L1": { + "current_price": stock_data["current_price"], + "foreigner_net": stock_data["foreigner_net"], + }, + "L2": { + "total_eval": total_eval, + "total_cash": total_cash, + "purchase_total": purchase_total, + "pnl_pct": pnl_pct, + }, + } + input_data = { + "current_price": stock_data["current_price"], + "foreigner_net": stock_data["foreigner_net"], + "total_eval": total_eval, + "total_cash": total_cash, + "pnl_pct": pnl_pct, + } + + decision_logger.log_decision( + stock_code=stock_code, + market=market.code, + exchange_code=market.exchange_code, + action=decision.action, + confidence=decision.confidence, + rationale=decision.rationale, + context_snapshot=context_snapshot, + input_data=input_data, + ) + + # Execute if actionable + if decision.action in ("BUY", "SELL"): + quantity = 1 + order_amount = stock_data["current_price"] * quantity + + # Risk check + 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) + continue # Skip this order + except CircuitBreakerTripped as exc: + logger.critical("Circuit breaker tripped — stopping session") + 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 + + # Send order + try: + if market.is_domestic: + result = await broker.send_order( + stock_code=stock_code, + order_type=decision.action, + quantity=quantity, + price=0, # market order + ) + else: + result = await overseas_broker.send_overseas_order( + exchange_code=market.exchange_code, + stock_code=stock_code, + order_type=decision.action, + quantity=quantity, + price=0.0, # market order + ) + logger.info("Order result: %s", result.get("msg1", "OK")) + + # Notify trade execution + try: + await telegram.notify_trade_execution( + stock_code=stock_code, + market=market.name, + action=decision.action, + quantity=quantity, + price=stock_data["current_price"], + confidence=decision.confidence, + ) + except Exception as exc: + logger.warning("Telegram notification failed: %s", exc) + except Exception as exc: + logger.error("Order execution failed for %s: %s", stock_code, exc) + continue + + # Log trade + log_trade( + conn=db_conn, + stock_code=stock_code, + action=decision.action, + confidence=decision.confidence, + rationale=decision.rationale, + market=market.code, + exchange_code=market.exchange_code, + ) + + logger.info("Daily trading session completed") + + async def run(settings: Settings) -> None: """Main async loop — iterate over open markets on a timer.""" broker = KISBroker(settings) @@ -375,7 +612,7 @@ async def run(settings: Settings) -> None: for sig in (signal.SIGINT, signal.SIGTERM): loop.add_signal_handler(sig, _signal_handler) - logger.info("The Ouroboros is alive. Mode: %s", settings.MODE) + logger.info("The Ouroboros is alive. Mode: %s, Trading: %s", settings.MODE, settings.TRADE_MODE) logger.info("Enabled markets: %s", settings.enabled_market_list) # Notify system startup @@ -385,170 +622,213 @@ async def run(settings: Settings) -> None: 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) + # Branch based on trading mode + if settings.TRADE_MODE == "daily": + # Daily trading mode: batch decisions at fixed intervals + logger.info( + "Daily trading mode: %d sessions every %d hours", + settings.DAILY_SESSIONS, + settings.SESSION_INTERVAL_HOURS, + ) - 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 + session_interval = settings.SESSION_INTERVAL_HOURS * 3600 # Convert to seconds - 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 + while not shutdown.is_set(): try: - next_market, next_open_time = get_next_market_open( - settings.enabled_market_list + await run_daily_session( + broker, + overseas_broker, + brain, + risk, + db_conn, + decision_logger, + context_store, + criticality_assessor, + telegram, + settings, ) - now = datetime.now(UTC) - wait_seconds = (next_open_time - now).total_seconds() - logger.info( - "No markets open. Next market: %s, opens in %.1f hours", - next_market.name, - wait_seconds / 3600, - ) - await asyncio.wait_for(shutdown.wait(), timeout=wait_seconds) - except TimeoutError: - continue # Market should be open now - except ValueError as exc: - logger.error("Failed to find next market open: %s", exc) - await asyncio.sleep(TRADE_INTERVAL_SECONDS) - continue - - # Process each open market - for market in open_markets: - if shutdown.is_set(): + except CircuitBreakerTripped: + logger.critical("Circuit breaker tripped — shutting down") + shutdown.set() break + except Exception as exc: + logger.exception("Daily session error: %s", exc) - # Notify market open if it just opened - if not _market_states.get(market.code, False): + # Wait for next session or shutdown + logger.info("Next session in %.1f hours", session_interval / 3600) + try: + await asyncio.wait_for(shutdown.wait(), timeout=session_interval) + except TimeoutError: + pass # Normal — time for next session + + else: + # Realtime trading mode: original per-stock loop + logger.info("Realtime trading mode: 60s interval per stock") + + 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: - 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) - if now_timestamp - last_scan >= SCAN_INTERVAL_SECONDS: - try: - # Scan all stocks in the universe - stock_universe = STOCK_UNIVERSE.get(market.code, []) - if stock_universe: - logger.info("Volatility Hunter: Scanning %s market", market.name) - scan_result = await market_scanner.scan_market( - market, stock_universe - ) - - # Update watchlist with top movers - current_watchlist = WATCHLISTS.get(market.code, []) - updated_watchlist = market_scanner.get_updated_watchlist( - current_watchlist, - scan_result, - max_replacements=2, - ) - WATCHLISTS[market.code] = updated_watchlist - - logger.info( - "Volatility Hunter: Watchlist updated for %s (%d top movers, %d breakouts)", - market.name, - len(scan_result.top_movers), - len(scan_result.breakouts), - ) - - last_scan_time[market.code] = now_timestamp - except Exception as exc: - logger.error("Volatility Hunter scan failed for %s: %s", market.name, exc) - - # Get watchlist for this market - watchlist = WATCHLISTS.get(market.code, []) - if not watchlist: - logger.debug("No watchlist for market %s", market.code) + next_market, next_open_time = get_next_market_open( + settings.enabled_market_list + ) + now = datetime.now(UTC) + wait_seconds = (next_open_time - now).total_seconds() + logger.info( + "No markets open. Next market: %s, opens in %.1f hours", + next_market.name, + wait_seconds / 3600, + ) + await asyncio.wait_for(shutdown.wait(), timeout=wait_seconds) + except TimeoutError: + continue # Market should be open now + except ValueError as exc: + logger.error("Failed to find next market open: %s", exc) + await asyncio.sleep(TRADE_INTERVAL_SECONDS) continue - logger.info("Processing market: %s (%d stocks)", market.name, len(watchlist)) - - # Process each stock in the watchlist - for stock_code in watchlist: + # Process each open market + for market in open_markets: if shutdown.is_set(): break - # Retry logic for connection errors - for attempt in range(1, MAX_CONNECTION_RETRIES + 1): + # Notify market open if it just opened + if not _market_states.get(market.code, False): try: - await trading_cycle( - broker, - overseas_broker, - brain, - risk, - db_conn, - decision_logger, - context_store, - criticality_assessor, - telegram, - market, - stock_code, - ) - break # Success — exit retry loop - 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: - logger.warning( - "Connection error for %s (attempt %d/%d): %s", - stock_code, - attempt, - MAX_CONNECTION_RETRIES, - exc, - ) - await asyncio.sleep(2**attempt) # Exponential backoff - else: - logger.error( - "Connection error for %s (all retries exhausted): %s", - stock_code, - exc, - ) - break # Give up on this stock + await telegram.notify_market_open(market.name) except Exception as exc: - logger.exception("Unexpected error for %s: %s", stock_code, exc) - break # Don't retry on unexpected errors + logger.warning("Market open notification failed: %s", exc) + _market_states[market.code] = True - # Log priority queue metrics periodically - metrics = await priority_queue.get_metrics() - if metrics.total_enqueued > 0: - logger.info( - "Priority queue metrics: enqueued=%d, dequeued=%d, size=%d, timeouts=%d, errors=%d", - metrics.total_enqueued, - metrics.total_dequeued, - metrics.current_size, - metrics.total_timeouts, - metrics.total_errors, - ) + # 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) + if now_timestamp - last_scan >= SCAN_INTERVAL_SECONDS: + try: + # Scan all stocks in the universe + stock_universe = STOCK_UNIVERSE.get(market.code, []) + if stock_universe: + logger.info("Volatility Hunter: Scanning %s market", market.name) + scan_result = await market_scanner.scan_market( + market, stock_universe + ) - # Wait for next cycle or shutdown - try: - await asyncio.wait_for(shutdown.wait(), timeout=TRADE_INTERVAL_SECONDS) - except TimeoutError: - pass # Normal — timeout means it's time for next cycle + # Update watchlist with top movers + current_watchlist = WATCHLISTS.get(market.code, []) + updated_watchlist = market_scanner.get_updated_watchlist( + current_watchlist, + scan_result, + max_replacements=2, + ) + WATCHLISTS[market.code] = updated_watchlist + + logger.info( + "Volatility Hunter: Watchlist updated for %s (%d top movers, %d breakouts)", + market.name, + len(scan_result.top_movers), + len(scan_result.breakouts), + ) + + last_scan_time[market.code] = now_timestamp + except Exception as exc: + logger.error("Volatility Hunter scan failed for %s: %s", market.name, exc) + + # Get watchlist for this market + watchlist = WATCHLISTS.get(market.code, []) + if not watchlist: + logger.debug("No watchlist for market %s", market.code) + continue + + logger.info("Processing market: %s (%d stocks)", market.name, len(watchlist)) + + # Process each stock in the watchlist + for stock_code in watchlist: + if shutdown.is_set(): + break + + # Retry logic for connection errors + for attempt in range(1, MAX_CONNECTION_RETRIES + 1): + try: + await trading_cycle( + broker, + overseas_broker, + brain, + risk, + db_conn, + decision_logger, + context_store, + criticality_assessor, + telegram, + market, + stock_code, + ) + break # Success — exit retry loop + 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: + logger.warning( + "Connection error for %s (attempt %d/%d): %s", + stock_code, + attempt, + MAX_CONNECTION_RETRIES, + exc, + ) + await asyncio.sleep(2**attempt) # Exponential backoff + else: + logger.error( + "Connection error for %s (all retries exhausted): %s", + stock_code, + exc, + ) + break # Give up on this stock + except Exception as exc: + logger.exception("Unexpected error for %s: %s", stock_code, exc) + break # Don't retry on unexpected errors + + # Log priority queue metrics periodically + metrics = await priority_queue.get_metrics() + if metrics.total_enqueued > 0: + logger.info( + "Priority queue metrics: enqueued=%d, dequeued=%d, size=%d, timeouts=%d, errors=%d", + metrics.total_enqueued, + metrics.total_dequeued, + metrics.current_size, + metrics.total_timeouts, + metrics.total_errors, + ) + + # Wait for next cycle or shutdown + try: + await asyncio.wait_for(shutdown.wait(), timeout=TRADE_INTERVAL_SECONDS) + except TimeoutError: + pass # Normal — timeout means it's time for next cycle finally: # Clean up resources await broker.close() diff --git a/tests/test_brain.py b/tests/test_brain.py index a6fe8d9..d6e67cc 100644 --- a/tests/test_brain.py +++ b/tests/test_brain.py @@ -152,3 +152,121 @@ class TestPromptConstruction: assert "JSON" in prompt assert "action" in prompt assert "confidence" in prompt + + +# --------------------------------------------------------------------------- +# Batch Decision Making +# --------------------------------------------------------------------------- + + +class TestBatchDecisionParsing: + """Batch response parser must handle JSON arrays correctly.""" + + def test_parse_valid_batch_response(self, settings): + client = GeminiClient(settings) + stocks_data = [ + {"stock_code": "AAPL", "current_price": 185.5}, + {"stock_code": "MSFT", "current_price": 420.0}, + ] + raw = """[ + {"code": "AAPL", "action": "BUY", "confidence": 85, "rationale": "Strong momentum"}, + {"code": "MSFT", "action": "HOLD", "confidence": 50, "rationale": "Wait for earnings"} + ]""" + + decisions = client._parse_batch_response(raw, stocks_data, token_count=100) + + assert len(decisions) == 2 + assert decisions["AAPL"].action == "BUY" + assert decisions["AAPL"].confidence == 85 + assert decisions["MSFT"].action == "HOLD" + assert decisions["MSFT"].confidence == 50 + + def test_parse_batch_with_markdown_wrapper(self, settings): + client = GeminiClient(settings) + stocks_data = [{"stock_code": "AAPL", "current_price": 185.5}] + raw = """```json +[{"code": "AAPL", "action": "BUY", "confidence": 90, "rationale": "Good"}] +```""" + + decisions = client._parse_batch_response(raw, stocks_data, token_count=100) + + assert decisions["AAPL"].action == "BUY" + assert decisions["AAPL"].confidence == 90 + + def test_parse_batch_empty_response_returns_hold_for_all(self, settings): + client = GeminiClient(settings) + stocks_data = [ + {"stock_code": "AAPL", "current_price": 185.5}, + {"stock_code": "MSFT", "current_price": 420.0}, + ] + + decisions = client._parse_batch_response("", stocks_data, token_count=100) + + assert len(decisions) == 2 + assert decisions["AAPL"].action == "HOLD" + assert decisions["AAPL"].confidence == 0 + assert decisions["MSFT"].action == "HOLD" + + def test_parse_batch_malformed_json_returns_hold_for_all(self, settings): + client = GeminiClient(settings) + stocks_data = [{"stock_code": "AAPL", "current_price": 185.5}] + raw = "This is not JSON" + + decisions = client._parse_batch_response(raw, stocks_data, token_count=100) + + assert decisions["AAPL"].action == "HOLD" + assert decisions["AAPL"].confidence == 0 + + def test_parse_batch_not_array_returns_hold_for_all(self, settings): + client = GeminiClient(settings) + stocks_data = [{"stock_code": "AAPL", "current_price": 185.5}] + raw = '{"code": "AAPL", "action": "BUY", "confidence": 90, "rationale": "Good"}' + + decisions = client._parse_batch_response(raw, stocks_data, token_count=100) + + assert decisions["AAPL"].action == "HOLD" + assert decisions["AAPL"].confidence == 0 + + def test_parse_batch_missing_stock_gets_hold(self, settings): + client = GeminiClient(settings) + stocks_data = [ + {"stock_code": "AAPL", "current_price": 185.5}, + {"stock_code": "MSFT", "current_price": 420.0}, + ] + # Response only has AAPL, MSFT is missing + raw = '[{"code": "AAPL", "action": "BUY", "confidence": 85, "rationale": "Good"}]' + + decisions = client._parse_batch_response(raw, stocks_data, token_count=100) + + assert decisions["AAPL"].action == "BUY" + assert decisions["MSFT"].action == "HOLD" + assert decisions["MSFT"].confidence == 0 + + def test_parse_batch_invalid_action_becomes_hold(self, settings): + client = GeminiClient(settings) + stocks_data = [{"stock_code": "AAPL", "current_price": 185.5}] + raw = '[{"code": "AAPL", "action": "YOLO", "confidence": 90, "rationale": "Moon"}]' + + decisions = client._parse_batch_response(raw, stocks_data, token_count=100) + + assert decisions["AAPL"].action == "HOLD" + + def test_parse_batch_low_confidence_becomes_hold(self, settings): + client = GeminiClient(settings) + stocks_data = [{"stock_code": "AAPL", "current_price": 185.5}] + raw = '[{"code": "AAPL", "action": "BUY", "confidence": 65, "rationale": "Weak"}]' + + decisions = client._parse_batch_response(raw, stocks_data, token_count=100) + + assert decisions["AAPL"].action == "HOLD" + assert decisions["AAPL"].confidence == 65 + + def test_parse_batch_missing_fields_gets_hold(self, settings): + client = GeminiClient(settings) + stocks_data = [{"stock_code": "AAPL", "current_price": 185.5}] + raw = '[{"code": "AAPL", "action": "BUY"}]' # Missing confidence and rationale + + decisions = client._parse_batch_response(raw, stocks_data, token_count=100) + + assert decisions["AAPL"].action == "HOLD" + assert decisions["AAPL"].confidence == 0