"""Signal, opportunity, and portfolio service tests.""" from __future__ import annotations import unittest from unittest.mock import patch from coinhunter.services import ( opportunity_service, portfolio_service, research_service, signal_service, ) class FakeSpotClient: def account_info(self): return { "balances": [ {"asset": "USDT", "free": "50", "locked": "0"}, {"asset": "BTC", "free": "0.01", "locked": "0"}, {"asset": "ETH", "free": "0.5", "locked": "0"}, {"asset": "DOGE", "free": "1", "locked": "0"}, ] } def ticker_price(self, symbols=None): mapping = { "BTCUSDT": {"symbol": "BTCUSDT", "price": "60000"}, "ETHUSDT": {"symbol": "ETHUSDT", "price": "3000"}, "DOGEUSDT": {"symbol": "DOGEUSDT", "price": "0.1"}, } return [mapping[symbol] for symbol in symbols] def ticker_stats(self, symbols=None, *, window="1d"): rows = { "BTCUSDT": { "symbol": "BTCUSDT", "lastPrice": "60000", "priceChangePercent": "5", "quoteVolume": "9000000", "highPrice": "60200", "lowPrice": "55000", }, "ETHUSDT": { "symbol": "ETHUSDT", "lastPrice": "3000", "priceChangePercent": "3", "quoteVolume": "8000000", "highPrice": "3100", "lowPrice": "2800", }, "SOLUSDT": { "symbol": "SOLUSDT", "lastPrice": "150", "priceChangePercent": "8", "quoteVolume": "10000000", "highPrice": "152", "lowPrice": "130", }, "DOGEUSDT": { "symbol": "DOGEUSDT", "lastPrice": "0.1", "priceChangePercent": "1", "quoteVolume": "100", "highPrice": "0.11", "lowPrice": "0.09", }, } if not symbols: return list(rows.values()) return [rows[symbol] for symbol in symbols] def exchange_info(self): return { "symbols": [ {"symbol": "BTCUSDT", "status": "TRADING"}, {"symbol": "ETHUSDT", "status": "TRADING"}, {"symbol": "SOLUSDT", "status": "TRADING"}, {"symbol": "DOGEUSDT", "status": "TRADING"}, ] } def klines(self, symbol, interval, limit): curves = { "BTCUSDT": [50000, 52000, 54000, 56000, 58000, 59000, 60000], "ETHUSDT": [2600, 2650, 2700, 2800, 2900, 2950, 3000], "SOLUSDT": [120, 125, 130, 135, 140, 145, 150], "DOGEUSDT": [0.11, 0.108, 0.105, 0.103, 0.102, 0.101, 0.1], }[symbol] rows = [] for index, close in enumerate(curves[-limit:]): rows.append( [ index, close * 0.98, close * 1.01, close * 0.97, close, 100 + index * 10, index + 1, close * (100 + index * 10), ] ) return rows class DustOverlapSpotClient(FakeSpotClient): def account_info(self): return {"balances": [{"asset": "XRP", "free": "5", "locked": "0"}]} def ticker_price(self, symbols=None): mapping = {"XRPUSDT": {"symbol": "XRPUSDT", "price": "1.5"}} return [mapping[symbol] for symbol in symbols] def ticker_stats(self, symbols=None, *, window="1d"): rows = { "XRPUSDT": { "symbol": "XRPUSDT", "lastPrice": "1.5", "priceChangePercent": "10", "quoteVolume": "5000000", "highPrice": "1.52", "lowPrice": "1.2", } } if not symbols: return list(rows.values()) return [rows[symbol] for symbol in symbols] def exchange_info(self): return {"symbols": [{"symbol": "XRPUSDT", "status": "TRADING"}]} def klines(self, symbol, interval, limit): rows = [] setup_curve = [ 1.4151, 1.4858, 1.3868, 1.5, 1.4009, 1.5142, 1.4151, 1.5, 1.4292, 1.4858, 1.4434, 1.4717, 1.4505, 1.4575, 1.4547, 1.4604, 1.4575, 1.4632, 1.4599, 1.466, 1.4618, 1.4698, 1.4745, 1.5, ] for index, close in enumerate(setup_curve[-limit:]): rows.append([index, close * 0.98, close * 1.01, close * 0.97, close, 100 + index * 10, index + 1, close * 100]) return rows class OpportunityPatternSpotClient: def account_info(self): return {"balances": [{"asset": "USDT", "free": "100", "locked": "0"}]} def ticker_price(self, symbols=None): return [] def ticker_stats(self, symbols=None, *, window="1d"): rows = { "SETUPUSDT": { "symbol": "SETUPUSDT", "lastPrice": "106", "priceChangePercent": "4", "quoteVolume": "10000000", "highPrice": "107", "lowPrice": "98", }, "CHASEUSDT": { "symbol": "CHASEUSDT", "lastPrice": "150", "priceChangePercent": "18", "quoteVolume": "9000000", "highPrice": "152", "lowPrice": "120", }, } if not symbols: return list(rows.values()) return [rows[symbol] for symbol in symbols] def exchange_info(self): return { "symbols": [ {"symbol": "SETUPUSDT", "status": "TRADING"}, {"symbol": "CHASEUSDT", "status": "TRADING"}, ] } def klines(self, symbol, interval, limit): curves = { "SETUPUSDT": [ 100, 105, 98, 106, 99, 107, 100, 106, 101, 105, 102, 104, 102.5, 103, 102.8, 103.2, 103.0, 103.4, 103.1, 103.6, 103.3, 103.8, 104.2, 106, ], "CHASEUSDT": [120, 125, 130, 135, 140, 145, 150], }[symbol] rows = [] for index, close in enumerate(curves[-limit:]): rows.append([index, close * 0.98, close * 1.01, close * 0.97, close, 100 + index * 20, index + 1, close * 100]) return rows class OpportunityServiceTestCase(unittest.TestCase): def setUp(self): self.config = { "market": {"default_quote": "USDT", "universe_allowlist": [], "universe_denylist": []}, "trading": {"dust_usdt_threshold": 10.0}, "signal": { "lookback_interval": "1h", "trend": 1.0, "momentum": 1.0, "breakout": 0.8, "volume": 0.7, "volatility_penalty": 0.5, }, "opportunity": { "scan_limit": 10, "top_n": 5, "min_quote_volume": 1000.0, "entry_threshold": 1.5, "watch_threshold": 0.6, "overlap_penalty": 0.6, "auto_research": False, "research_provider": "coingecko", "research_timeout_seconds": 4.0, "risk_limits": { "min_liquidity": 0.0, "max_overextension": 0.08, "max_downside_risk": 0.3, "max_unlock_risk": 0.75, "max_regulatory_risk": 0.75, "min_quality_for_add": 0.0, }, "weights": { "trend": 1.0, "momentum": 1.0, "breakout": 0.8, "pullback": 0.4, "volume": 0.7, "liquidity": 0.3, "trend_alignment": 0.8, "fundamental": 0.8, "tokenomics": 0.7, "catalyst": 0.5, "adoption": 0.4, "smart_money": 0.3, "volatility_penalty": 0.5, "overextension_penalty": 0.7, "downside_penalty": 0.5, "unlock_penalty": 0.8, "regulatory_penalty": 0.4, "position_concentration_penalty": 0.6, }, }, "portfolio": { "add_threshold": 1.5, "hold_threshold": 0.6, "trim_threshold": 0.2, "exit_threshold": -0.2, "max_position_weight": 0.6, }, } def test_portfolio_analysis_ignores_dust_and_emits_recommendations(self): events = [] with patch.object(portfolio_service, "audit_event", side_effect=lambda event, payload, **kwargs: events.append(event)): payload = portfolio_service.analyze_portfolio(self.config, spot_client=FakeSpotClient()) symbols = [item["symbol"] for item in payload["recommendations"]] self.assertNotIn("DOGEUSDT", symbols) self.assertEqual(symbols, ["BTCUSDT", "ETHUSDT"]) self.assertEqual(payload["recommendations"][0]["action"], "add") self.assertEqual(payload["recommendations"][1]["action"], "hold") self.assertEqual(events, ["opportunity_portfolio_generated"]) def test_scan_is_deterministic(self): with patch.object(opportunity_service, "audit_event", return_value=None): payload = opportunity_service.scan_opportunities( self.config | {"opportunity": self.config["opportunity"] | {"top_n": 2}}, spot_client=OpportunityPatternSpotClient(), ) self.assertEqual([item["symbol"] for item in payload["recommendations"]], ["SETUPUSDT", "CHASEUSDT"]) self.assertEqual([item["action"] for item in payload["recommendations"]], ["trigger", "chase"]) self.assertGreater(payload["recommendations"][0]["metrics"]["setup_score"], 0.6) self.assertGreater(payload["recommendations"][1]["metrics"]["extension_penalty"], 1.0) def test_scan_respects_ignore_dust_for_overlap_penalty(self): client = DustOverlapSpotClient() base_config = self.config | { "opportunity": self.config["opportunity"] | { "top_n": 1, "ignore_dust": True, "overlap_penalty": 2.0, } } with patch.object(opportunity_service, "audit_event", return_value=None): ignored = opportunity_service.scan_opportunities(base_config, spot_client=client, symbols=["XRPUSDT"]) included = opportunity_service.scan_opportunities( base_config | {"opportunity": base_config["opportunity"] | {"ignore_dust": False}}, spot_client=client, symbols=["XRPUSDT"], ) ignored_rec = ignored["recommendations"][0] included_rec = included["recommendations"][0] self.assertEqual(ignored_rec["action"], "trigger") self.assertEqual(ignored_rec["metrics"]["position_weight"], 0.0) self.assertEqual(included_rec["action"], "skip") self.assertEqual(included_rec["metrics"]["position_weight"], 1.0) self.assertLess(included_rec["score"], ignored_rec["score"]) def test_signal_score_handles_empty_klines(self): score, metrics = signal_service.score_market_signal([], [], {"price_change_pct": 1.0}, {}) self.assertEqual(score, 0.0) self.assertEqual(metrics["trend"], 0.0) def test_overextended_candidate_is_not_an_add(self): closes = [100, 110, 121, 133, 146, 160, 176] volumes = [100, 120, 130, 150, 170, 190, 230] ticker = { "price_change_pct": 35.0, "quote_volume": 20_000_000.0, "high_price": 180.0, "low_price": 95.0, } score, metrics = opportunity_service._score_candidate( closes, volumes, ticker, self.config["opportunity"]["weights"], 0.0, {"1h": closes, "4h": closes} ) action, reasons = opportunity_service._action_for(score, 0.0, metrics) self.assertGreater(score, 1.0) self.assertGreater(metrics["overextension"], 0.08) self.assertEqual(action, "observe") self.assertIn("move looks extended; wait for a cleaner entry", reasons) def test_external_research_signals_improve_candidate_quality(self): closes = [100, 101, 102, 103, 104, 105, 106] volumes = [100, 105, 110, 115, 120, 125, 130] ticker = { "price_change_pct": 4.0, "quote_volume": 50_000_000.0, "high_price": 110.0, "low_price": 95.0, } base_score, base_metrics = opportunity_service._score_candidate( closes, volumes, ticker, self.config["opportunity"]["weights"], 0.0, {"1h": closes} ) researched_score, researched_metrics = opportunity_service._score_candidate( closes, volumes, ticker, self.config["opportunity"]["weights"], 0.0, {"1h": closes}, { "fundamental": 85, "tokenomics": 80, "catalyst": 70, "adoption": 90, "smart_money": 60, }, ) self.assertGreater(researched_score, base_score) self.assertEqual(base_metrics["quality"], 0.0) self.assertGreater(researched_metrics["quality"], 0.7) def test_scan_uses_automatic_external_research(self): config = self.config | { "opportunity": self.config["opportunity"] | { "auto_research": True, "top_n": 2, } } with ( patch.object(opportunity_service, "audit_event", return_value=None), patch.object( opportunity_service, "get_external_research", return_value={ "SOLUSDT": { "fundamental": 0.9, "tokenomics": 0.8, "catalyst": 0.9, "adoption": 0.8, "smart_money": 0.7, "unlock_risk": 0.1, "regulatory_risk": 0.1, "research_confidence": 0.9, } }, ) as research_mock, ): payload = opportunity_service.scan_opportunities(config, spot_client=FakeSpotClient()) research_mock.assert_called_once() sol = next(item for item in payload["recommendations"] if item["symbol"] == "SOLUSDT") self.assertEqual(sol["metrics"]["fundamental"], 0.9) self.assertEqual(sol["metrics"]["research_confidence"], 0.9) def test_research_score_does_not_create_weak_trigger(self): metrics = { "extension_penalty": 0.0, "recent_runup": 0.0, "breakout_pct": -0.01, "setup_score": 0.12, "trigger_score": 0.18, } action, reasons = opportunity_service._action_for_opportunity( 2.5, metrics, { "entry_threshold": 1.5, "watch_threshold": 0.6, "min_trigger_score": 0.45, "min_setup_score": 0.35, }, ) self.assertEqual(action, "setup") self.assertIn("technical trigger quality is not clean enough", reasons[0]) def test_unlock_risk_blocks_add_recommendation(self): metrics = { "liquidity": 0.8, "overextension": 0.0, "downside_risk": 0.0, "unlock_risk": 0.9, "regulatory_risk": 0.0, "quality": 0.8, } action, reasons = opportunity_service._action_for( 3.0, 0.0, metrics, self.config["opportunity"]["risk_limits"], ) self.assertEqual(action, "observe") self.assertIn("token unlock or dilution risk is too high", reasons) class ResearchServiceTestCase(unittest.TestCase): def test_coingecko_market_data_becomes_research_signals(self): signals = research_service._coingecko_market_to_signals( { "id": "solana", "symbol": "sol", "market_cap": 80_000_000_000, "fully_diluted_valuation": 95_000_000_000, "total_volume": 5_000_000_000, "market_cap_rank": 6, "circulating_supply": 550_000_000, "total_supply": 600_000_000, "max_supply": None, "price_change_percentage_7d_in_currency": 12.0, "price_change_percentage_30d_in_currency": 35.0, "price_change_percentage_200d_in_currency": 80.0, }, is_trending=True, ) self.assertGreater(signals["fundamental"], 0.6) self.assertGreater(signals["tokenomics"], 0.8) self.assertGreater(signals["catalyst"], 0.6) self.assertLess(signals["unlock_risk"], 0.2)