317 lines
11 KiB
Python
317 lines
11 KiB
Python
"""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, 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,
|
|
},
|
|
"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)
|