import asyncio import unittest from datetime import datetime import pandas as pd from upbit_trader import ( UpbitTrader, analyze_order_flow, check_trend, get_entry_trend_adjustment, rank_market_tickers, reconstruct_virtual_account, ) class TrendStrategyTests(unittest.TestCase): def test_volume_ranking_keeps_the_entire_market(self): tickers = [f"KRW-C{i:03d}" for i in range(275)] volumes = {ticker: index for index, ticker in enumerate(tickers)} ranked = rank_market_tickers(tickers, volumes) self.assertEqual(len(ranked), 275) self.assertEqual(set(ranked), set(tickers)) self.assertEqual(ranked[0], "KRW-C274") def test_btc_downtrend_is_a_penalty_not_a_global_block(self): self.assertEqual( get_entry_trend_adjustment("KRW-ONDO", "up", "down"), 0, ) def test_major_coin_requires_btc_uptrend(self): self.assertIsNone( get_entry_trend_adjustment("KRW-ETH", "up", "neutral") ) self.assertIsNone( get_entry_trend_adjustment("KRW-ETH", "up", "down") ) def test_non_rising_asset_is_still_blocked(self): self.assertIsNone( get_entry_trend_adjustment("KRW-ETH", "neutral", "up") ) def test_btc_does_not_receive_its_trend_bonus_twice(self): self.assertEqual( get_entry_trend_adjustment("KRW-BTC", "up", "up"), 2, ) def test_downtrend_has_finite_risk_penalty(self): close = pd.Series(range(160, 100, -1), dtype=float) trend, score = check_trend(pd.DataFrame({"close": close})) self.assertEqual(trend, "down") self.assertEqual(score, -2) class OrderFlowTests(unittest.TestCase): @staticmethod def _orderbook(bid_size, ask_size, spread=1): return { "orderbook_units": [ { "bid_price": 100_000, "bid_size": bid_size, "ask_price": 100_000 + spread, "ask_size": ask_size, } for _ in range(5) ] } @staticmethod def _trades(buy_count, sell_count, buy_size=2, sell_size=1): return [ {"ask_bid": "BID", "trade_price": 100_000, "trade_volume": buy_size} for _ in range(buy_count) ] + [ {"ask_bid": "ASK", "trade_price": 100_000, "trade_volume": sell_size} for _ in range(sell_count) ] def test_scores_combined_buy_pressure(self): result = analyze_order_flow( self._orderbook(bid_size=4, ask_size=1), self._trades(buy_count=40, sell_count=10), ) self.assertIsNotNone(result) self.assertEqual(result["score"], 3) self.assertFalse(result["blocked"]) def test_blocks_combined_sell_pressure(self): result = analyze_order_flow( self._orderbook(bid_size=1, ask_size=4), self._trades( buy_count=10, sell_count=40, buy_size=1, sell_size=2, ), ) self.assertIsNotNone(result) self.assertLessEqual(result["score"], -2) self.assertTrue(result["blocked"]) def test_blocks_excessive_spread(self): result = analyze_order_flow( self._orderbook(bid_size=4, ask_size=1, spread=1), self._trades(buy_count=40, sell_count=10), ) self.assertFalse(result["blocked"]) wide_result = analyze_order_flow( self._orderbook(bid_size=4, ask_size=1, spread=400), self._trades(buy_count=40, sell_count=10), ) self.assertTrue(wide_result["blocked"]) class TradingNotificationTests(unittest.TestCase): def test_reconstructs_legacy_virtual_cash_from_profit(self): cash, coins = reconstruct_virtual_account( { "total_budget": 10_000_000, "total_realized": 60_078, "virtual_krw": None, }, [], ) self.assertEqual(cash, 10_060_078) self.assertEqual(coins, {}) def test_reconstructs_legacy_virtual_position_and_debited_cash(self): cash, coins = reconstruct_virtual_account( { "total_budget": 1_000_000, "total_realized": 10_000, "virtual_krw": None, }, [{ "coin": "UNI", "status": "holding", "buy_price": 5_000, "buy_volume": 99.95, }], ) self.assertAlmostEqual(cash, 510_000) self.assertEqual(coins, {"UNI": 99.95}) def test_prefers_persisted_virtual_cash(self): cash, coins = reconstruct_virtual_account( { "total_budget": 1_000_000, "total_realized": 10_000, "virtual_krw": 321_000, }, [{ "coin": "UNI", "status": "holding", "buy_price": 5_000, "buy_volume": 99.95, }], ) self.assertEqual(cash, 321_000) self.assertEqual(coins, {"UNI": 99.95}) def test_daily_screen_cache_is_reused_within_ttl(self): async def check_cache(): trader = UpbitTrader() trader._daily_screen_cache = { "KRW-BTC": {"trend": "up", "daily_atr_pct": 2.0} } trader._daily_screen_cache_time = asyncio.get_event_loop().time() return await trader._ensure_daily_screen_cache(["KRW-BTC"]) self.assertEqual(asyncio.run(check_cache()), (False, 0)) def test_immediate_rescan_clears_closed_candle_gate(self): trader = UpbitTrader() trader._last_scan = 123 trader._last_scan_candle = datetime(2026, 7, 30, 10, 0) trader.request_immediate_rescan() self.assertEqual(trader._last_scan, 0) self.assertIsNone(trader._last_scan_candle) def test_status_exposes_last_scan_breakdown(self): trader = UpbitTrader() trader.running = True trader.paper_mode = True trader.total_budget = 1_000_000 trader._virtual_krw = 1_000_000 trader._market_trend = "down" trader._last_scan_at = datetime(2026, 7, 30, 10, 15, 23) trader._last_scan_candle = datetime(2026, 7, 30, 10, 0) trader._last_scan_stats = { "scanned": 80, "candidates": 0, "trend_down": 55, "trend_neutral": 7, "daily_data": 7, "daily_atr": 5, "signal_filter": 6, } trader._last_scan_analyzed = ["ONDO", "SHIB", "SOON"] status = asyncio.run(trader.get_status_text()) self.assertIn("최근 스캔: 07-30 10:15:23 | 시장기준 BTC down", status) self.assertIn("스캔 결과: 후보 0 / 80종목", status) self.assertIn("추세/시장 62", status) self.assertIn("15분신호 6", status) self.assertIn("15분 분석 종목: ONDO, SHIB, SOON", status) self.assertIn("알트코인은 각각 독립 분석", status) self.assertIn("다음 스캔:", status) def test_empty_market_scan_does_not_send_message(self): async def run_cycle(): trader = UpbitTrader() trader.running = True trader.paper_mode = True trader.total_budget = 100_000 trader._virtual_krw = 100_000 async def empty_scan(): trader._market_trend = "down" trader._strategy = "BTC down | 상대강도 신호 대기" return [] messages = [] async def send_message(message): messages.append(message) trader.scan_market = empty_scan await trader.execute_cycle(send_message) return messages self.assertEqual(asyncio.run(run_cycle()), []) if __name__ == "__main__": unittest.main()