feat: add Polymarket API tool functions with tests
Implements 11 @tool-decorated functions for Polymarket prediction market data access (Gamma, CLOB, Data APIs), social sentiment, and news search, plus 48 unit tests covering success paths, edge cases, and error handling. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
test
parent
50a705dc12
commit
12051c1570
|
|
@ -0,0 +1,614 @@
|
|||
"""
|
||||
Tests for tradingagents.agents.utils.polymarket_tools
|
||||
|
||||
All HTTP calls are mocked with unittest.mock.patch so no live network access
|
||||
is required.
|
||||
"""
|
||||
|
||||
import json
|
||||
import unittest
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
from tradingagents.agents.utils.polymarket_tools import (
|
||||
_api_get,
|
||||
get_market_data,
|
||||
get_price_history,
|
||||
get_event_details,
|
||||
get_orderbook,
|
||||
get_event_news,
|
||||
get_global_news,
|
||||
get_whale_activity,
|
||||
get_market_stats,
|
||||
get_leaderboard_signals,
|
||||
get_social_sentiment,
|
||||
search_markets,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helper: build a minimal Gamma event payload
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _gamma_event():
|
||||
return {
|
||||
"id": "123",
|
||||
"title": "Test Election 2026",
|
||||
"slug": "test-election-2026",
|
||||
"active": True,
|
||||
"startDate": "2026-01-01T00:00:00Z",
|
||||
"endDate": "2026-12-31T00:00:00Z",
|
||||
"volume": "500000",
|
||||
"liquidity": "50000",
|
||||
"description": "Who will win the 2026 test election?",
|
||||
"resolutionSource": "https://example.com",
|
||||
"tags": [{"label": "Politics"}],
|
||||
"markets": [
|
||||
{
|
||||
"question": "Will Candidate A win?",
|
||||
"conditionId": "cond-001",
|
||||
"active": True,
|
||||
"endDate": "2026-12-31T00:00:00Z",
|
||||
"outcomePrices": json.dumps(["0.60", "0.40"]),
|
||||
"outcomes": json.dumps(["Yes", "No"]),
|
||||
"spread": "0.02",
|
||||
"volume": "300000",
|
||||
"description": "Resolves YES if Candidate A wins.",
|
||||
}
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _api_get
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestApiGet(unittest.TestCase):
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools.requests.get")
|
||||
def test_success(self, mock_get):
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.raise_for_status.return_value = None
|
||||
mock_resp.json.return_value = {"key": "value"}
|
||||
mock_get.return_value = mock_resp
|
||||
|
||||
result = _api_get("https://example.com/test")
|
||||
self.assertEqual(result, {"key": "value"})
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools.requests.get")
|
||||
@patch("tradingagents.agents.utils.polymarket_tools.time.sleep", return_value=None)
|
||||
def test_retry_then_raise(self, mock_sleep, mock_get):
|
||||
import requests as req
|
||||
mock_get.side_effect = req.ConnectionError("refused")
|
||||
with self.assertRaises(req.ConnectionError):
|
||||
_api_get("https://example.com/fail")
|
||||
self.assertEqual(mock_get.call_count, 3)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# get_market_data
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestGetMarketData(unittest.TestCase):
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_returns_string(self, mock_api):
|
||||
mock_api.return_value = _gamma_event()
|
||||
result = get_market_data.invoke({"event_id": "123"})
|
||||
self.assertIsInstance(result, str)
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_contains_title(self, mock_api):
|
||||
mock_api.return_value = _gamma_event()
|
||||
result = get_market_data.invoke({"event_id": "123"})
|
||||
self.assertIn("Test Election 2026", result)
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_contains_volume(self, mock_api):
|
||||
mock_api.return_value = _gamma_event()
|
||||
result = get_market_data.invoke({"event_id": "123"})
|
||||
self.assertIn("500000", result)
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_contains_outcome_prices(self, mock_api):
|
||||
mock_api.return_value = _gamma_event()
|
||||
result = get_market_data.invoke({"event_id": "123"})
|
||||
self.assertIn("Yes", result)
|
||||
self.assertIn("0.60", result)
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_error_handling(self, mock_api):
|
||||
mock_api.side_effect = Exception("network error")
|
||||
result = get_market_data.invoke({"event_id": "bad-id"})
|
||||
self.assertIsInstance(result, str)
|
||||
self.assertIn("Error", result)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# get_price_history
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestGetPriceHistory(unittest.TestCase):
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_returns_string(self, mock_api):
|
||||
mock_api.return_value = {
|
||||
"history": [
|
||||
{"t": "2026-01-01T00:00:00Z", "p": 0.55},
|
||||
{"t": "2026-01-02T00:00:00Z", "p": 0.60},
|
||||
{"t": "2026-01-03T00:00:00Z", "p": 0.58},
|
||||
]
|
||||
}
|
||||
result = get_price_history.invoke({"token_id": "tok-001"})
|
||||
self.assertIsInstance(result, str)
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_contains_stats(self, mock_api):
|
||||
mock_api.return_value = {
|
||||
"history": [
|
||||
{"t": "2026-01-01T00:00:00Z", "p": 0.5},
|
||||
{"t": "2026-01-02T00:00:00Z", "p": 0.7},
|
||||
]
|
||||
}
|
||||
result = get_price_history.invoke({"token_id": "tok-001"})
|
||||
self.assertIn("Latest Price", result)
|
||||
self.assertIn("Min Price", result)
|
||||
self.assertIn("Max Price", result)
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_empty_history(self, mock_api):
|
||||
mock_api.return_value = {"history": []}
|
||||
result = get_price_history.invoke({"token_id": "tok-001"})
|
||||
self.assertIn("No price history", result)
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_error_handling(self, mock_api):
|
||||
mock_api.side_effect = Exception("timeout")
|
||||
result = get_price_history.invoke({"token_id": "bad-token"})
|
||||
self.assertIsInstance(result, str)
|
||||
self.assertIn("Error", result)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# get_event_details
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestGetEventDetails(unittest.TestCase):
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_returns_string(self, mock_api):
|
||||
mock_api.return_value = _gamma_event()
|
||||
result = get_event_details.invoke({"event_id": "123"})
|
||||
self.assertIsInstance(result, str)
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_contains_description(self, mock_api):
|
||||
mock_api.return_value = _gamma_event()
|
||||
result = get_event_details.invoke({"event_id": "123"})
|
||||
self.assertIn("Who will win", result)
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_contains_resolution_deadline(self, mock_api):
|
||||
mock_api.return_value = _gamma_event()
|
||||
result = get_event_details.invoke({"event_id": "123"})
|
||||
self.assertIn("Resolution Deadline", result)
|
||||
self.assertIn("2026-12-31", result)
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_contains_market_question(self, mock_api):
|
||||
mock_api.return_value = _gamma_event()
|
||||
result = get_event_details.invoke({"event_id": "123"})
|
||||
self.assertIn("Will Candidate A win?", result)
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_error_handling(self, mock_api):
|
||||
mock_api.side_effect = Exception("404")
|
||||
result = get_event_details.invoke({"event_id": "missing"})
|
||||
self.assertIsInstance(result, str)
|
||||
self.assertIn("Error", result)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# get_orderbook
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestGetOrderbook(unittest.TestCase):
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_returns_string(self, mock_api):
|
||||
mock_api.return_value = {
|
||||
"market": "test-market",
|
||||
"asset_id": "tok-001",
|
||||
"bids": [{"price": "0.59", "size": "100"}, {"price": "0.58", "size": "200"}],
|
||||
"asks": [{"price": "0.61", "size": "150"}, {"price": "0.62", "size": "300"}],
|
||||
}
|
||||
result = get_orderbook.invoke({"token_id": "tok-001"})
|
||||
self.assertIsInstance(result, str)
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_contains_bids_and_asks(self, mock_api):
|
||||
mock_api.return_value = {
|
||||
"market": "test-market",
|
||||
"asset_id": "tok-001",
|
||||
"bids": [{"price": "0.59", "size": "100"}],
|
||||
"asks": [{"price": "0.61", "size": "150"}],
|
||||
}
|
||||
result = get_orderbook.invoke({"token_id": "tok-001"})
|
||||
self.assertIn("Bids", result)
|
||||
self.assertIn("Asks", result)
|
||||
self.assertIn("0.59", result)
|
||||
self.assertIn("0.61", result)
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_spread_calculation(self, mock_api):
|
||||
mock_api.return_value = {
|
||||
"market": "test-market",
|
||||
"asset_id": "tok-001",
|
||||
"bids": [{"price": "0.58", "size": "100"}],
|
||||
"asks": [{"price": "0.62", "size": "100"}],
|
||||
}
|
||||
result = get_orderbook.invoke({"token_id": "tok-001"})
|
||||
self.assertIn("Spread", result)
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_error_handling(self, mock_api):
|
||||
mock_api.side_effect = Exception("connection refused")
|
||||
result = get_orderbook.invoke({"token_id": "bad-token"})
|
||||
self.assertIsInstance(result, str)
|
||||
self.assertIn("Error", result)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# get_event_news
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestGetEventNews(unittest.TestCase):
|
||||
|
||||
def test_no_api_key_returns_message(self):
|
||||
import os
|
||||
os.environ.pop("TAVILY_API_KEY", None)
|
||||
result = get_event_news.invoke({"query": "test election"})
|
||||
self.assertIsInstance(result, str)
|
||||
self.assertIn("TAVILY_API_KEY", result)
|
||||
|
||||
def test_with_mocked_tavily(self):
|
||||
mock_client = MagicMock()
|
||||
mock_client.search.return_value = {
|
||||
"results": [
|
||||
{
|
||||
"title": "Election Update",
|
||||
"url": "https://news.example.com/1",
|
||||
"content": "Candidate A leads in polls.",
|
||||
"published_date": "2026-03-01",
|
||||
}
|
||||
]
|
||||
}
|
||||
mock_tavily_module = MagicMock()
|
||||
mock_tavily_module.TavilyClient.return_value = mock_client
|
||||
|
||||
with patch.dict("sys.modules", {"tavily": mock_tavily_module}):
|
||||
result = get_event_news.invoke({"query": "election 2026", "api_key": "fake-key"})
|
||||
|
||||
self.assertIsInstance(result, str)
|
||||
self.assertIn("Election Update", result)
|
||||
self.assertIn("election 2026", result)
|
||||
|
||||
def test_error_handling(self):
|
||||
mock_tavily_module = MagicMock()
|
||||
mock_tavily_module.TavilyClient.side_effect = Exception("API error")
|
||||
|
||||
with patch.dict("sys.modules", {"tavily": mock_tavily_module}):
|
||||
result = get_event_news.invoke({"query": "test", "api_key": "fake-key-for-error-test"})
|
||||
|
||||
self.assertIsInstance(result, str)
|
||||
self.assertIn("failed", result.lower())
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# get_global_news
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestGetGlobalNews(unittest.TestCase):
|
||||
|
||||
def test_no_api_key_returns_message(self):
|
||||
import os
|
||||
os.environ.pop("TAVILY_API_KEY", None)
|
||||
result = get_global_news.invoke({})
|
||||
self.assertIsInstance(result, str)
|
||||
self.assertIn("TAVILY_API_KEY", result)
|
||||
|
||||
def test_with_mocked_tavily(self):
|
||||
mock_client = MagicMock()
|
||||
mock_client.search.return_value = {
|
||||
"results": [
|
||||
{
|
||||
"title": "Global Markets Update",
|
||||
"url": "https://finance.example.com/1",
|
||||
"content": "Markets rose globally.",
|
||||
"published_date": "2026-03-20",
|
||||
}
|
||||
]
|
||||
}
|
||||
mock_tavily_module = MagicMock()
|
||||
mock_tavily_module.TavilyClient.return_value = mock_client
|
||||
|
||||
with patch.dict("sys.modules", {"tavily": mock_tavily_module}):
|
||||
result = get_global_news.invoke({"query": "global markets 2026", "api_key": "fake-key"})
|
||||
|
||||
self.assertIsInstance(result, str)
|
||||
self.assertIn("Global Markets Update", result)
|
||||
|
||||
def test_error_returns_string(self):
|
||||
mock_tavily_module = MagicMock()
|
||||
mock_tavily_module.TavilyClient.side_effect = RuntimeError("rate limit")
|
||||
|
||||
with patch.dict("sys.modules", {"tavily": mock_tavily_module}):
|
||||
result = get_global_news.invoke({"query": "test", "api_key": "fake-key-for-error-test"})
|
||||
|
||||
self.assertIsInstance(result, str)
|
||||
self.assertIn("failed", result.lower())
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# get_whale_activity
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestGetWhaleActivity(unittest.TestCase):
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_returns_string(self, mock_api):
|
||||
mock_api.return_value = [
|
||||
{"proxyWallet": "0xABCDEF123456", "position": "10000", "value": "6000"},
|
||||
{"proxyWallet": "0x789012345678", "position": "5000", "value": "3000"},
|
||||
]
|
||||
result = get_whale_activity.invoke({"market_id": "cond-001"})
|
||||
self.assertIsInstance(result, str)
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_contains_table_headers(self, mock_api):
|
||||
mock_api.return_value = [
|
||||
{"proxyWallet": "0xABCDEF1234567890", "position": "10000", "value": "6000"},
|
||||
]
|
||||
result = get_whale_activity.invoke({"market_id": "cond-001"})
|
||||
self.assertIn("Rank", result)
|
||||
self.assertIn("Address", result)
|
||||
self.assertIn("Position", result)
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_empty_holders(self, mock_api):
|
||||
mock_api.return_value = []
|
||||
result = get_whale_activity.invoke({"market_id": "cond-001"})
|
||||
self.assertIsInstance(result, str)
|
||||
self.assertIn("No holder data", result)
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_error_handling(self, mock_api):
|
||||
mock_api.side_effect = Exception("server error")
|
||||
result = get_whale_activity.invoke({"market_id": "bad-market"})
|
||||
self.assertIsInstance(result, str)
|
||||
self.assertIn("Error", result)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# get_market_stats
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestGetMarketStats(unittest.TestCase):
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_returns_string(self, mock_api):
|
||||
mock_api.return_value = {
|
||||
"openInterest": "250000",
|
||||
"totalVolume": "500000",
|
||||
"numTraders": "1234",
|
||||
"liquidity": "75000",
|
||||
"lastTradePrice": "0.62",
|
||||
}
|
||||
result = get_market_stats.invoke({"market_id": "cond-001"})
|
||||
self.assertIsInstance(result, str)
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_contains_open_interest(self, mock_api):
|
||||
mock_api.return_value = {
|
||||
"openInterest": "250000",
|
||||
"totalVolume": "500000",
|
||||
"numTraders": "1234",
|
||||
"liquidity": "75000",
|
||||
"lastTradePrice": "0.62",
|
||||
}
|
||||
result = get_market_stats.invoke({"market_id": "cond-001"})
|
||||
self.assertIn("Open Interest", result)
|
||||
self.assertIn("250000", result)
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_contains_volume(self, mock_api):
|
||||
mock_api.return_value = {
|
||||
"openInterest": "250000",
|
||||
"totalVolume": "500000",
|
||||
}
|
||||
result = get_market_stats.invoke({"market_id": "cond-001"})
|
||||
self.assertIn("Volume", result)
|
||||
self.assertIn("500000", result)
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_error_handling(self, mock_api):
|
||||
mock_api.side_effect = Exception("404 not found")
|
||||
result = get_market_stats.invoke({"market_id": "missing"})
|
||||
self.assertIsInstance(result, str)
|
||||
self.assertIn("Error", result)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# get_leaderboard_signals
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestGetLeaderboardSignals(unittest.TestCase):
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_returns_string(self, mock_api):
|
||||
mock_api.return_value = [
|
||||
{"name": "TraderAlpha", "pnl": "12000", "volume": "300000"},
|
||||
{"name": "TraderBeta", "pnl": "9000", "volume": "250000"},
|
||||
]
|
||||
result = get_leaderboard_signals.invoke({"category": "OVERALL", "time_period": "WEEK"})
|
||||
self.assertIsInstance(result, str)
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_contains_traders(self, mock_api):
|
||||
mock_api.return_value = [
|
||||
{"name": "TraderAlpha", "pnl": "12000", "volume": "300000"},
|
||||
]
|
||||
result = get_leaderboard_signals.invoke({"category": "OVERALL", "time_period": "WEEK"})
|
||||
self.assertIn("TraderAlpha", result)
|
||||
self.assertIn("12000", result)
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_empty_leaderboard(self, mock_api):
|
||||
mock_api.return_value = []
|
||||
result = get_leaderboard_signals.invoke({})
|
||||
self.assertIsInstance(result, str)
|
||||
self.assertIn("No leaderboard data", result)
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_error_handling(self, mock_api):
|
||||
mock_api.side_effect = Exception("unauthorized")
|
||||
result = get_leaderboard_signals.invoke({"category": "POLITICS", "time_period": "MONTH"})
|
||||
self.assertIsInstance(result, str)
|
||||
self.assertIn("Error", result)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# get_social_sentiment
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestGetSocialSentiment(unittest.TestCase):
|
||||
|
||||
def test_no_keys_graceful(self):
|
||||
import os
|
||||
for key in ("TWITTER_BEARER_TOKEN", "REDDIT_CLIENT_ID", "REDDIT_CLIENT_SECRET"):
|
||||
os.environ.pop(key, None)
|
||||
result = get_social_sentiment.invoke({"query": "polymarket election"})
|
||||
self.assertIsInstance(result, str)
|
||||
self.assertIn("skipping", result.lower())
|
||||
|
||||
def test_twitter_with_mock(self):
|
||||
mock_tweet = MagicMock()
|
||||
mock_tweet.text = "Candidate A is looking strong on Polymarket!"
|
||||
mock_tweet.public_metrics = {"like_count": 10, "retweet_count": 2}
|
||||
|
||||
mock_client = MagicMock()
|
||||
mock_client.search_recent_tweets.return_value = MagicMock(data=[mock_tweet])
|
||||
|
||||
mock_tweepy = MagicMock()
|
||||
mock_tweepy.Client.return_value = mock_client
|
||||
|
||||
with patch.dict("sys.modules", {"tweepy": mock_tweepy}):
|
||||
with patch.dict("os.environ", {"TWITTER_BEARER_TOKEN": "fake-token"}):
|
||||
result = get_social_sentiment.invoke({"query": "polymarket election"})
|
||||
|
||||
self.assertIsInstance(result, str)
|
||||
self.assertIn("Twitter", result)
|
||||
self.assertIn("Candidate A", result)
|
||||
|
||||
def test_reddit_with_mock(self):
|
||||
mock_post = MagicMock()
|
||||
mock_post.title = "Polymarket odds for election shifting"
|
||||
mock_post.score = 150
|
||||
mock_post.subreddit.display_name = "Polymarket"
|
||||
|
||||
mock_reddit_instance = MagicMock()
|
||||
mock_reddit_instance.subreddit.return_value.search.return_value = [mock_post]
|
||||
|
||||
mock_praw = MagicMock()
|
||||
mock_praw.Reddit.return_value = mock_reddit_instance
|
||||
|
||||
with patch.dict("sys.modules", {"praw": mock_praw}):
|
||||
with patch.dict("os.environ", {
|
||||
"REDDIT_CLIENT_ID": "fake-id",
|
||||
"REDDIT_CLIENT_SECRET": "fake-secret",
|
||||
}):
|
||||
# Remove Twitter token so we focus on Reddit
|
||||
import os
|
||||
os.environ.pop("TWITTER_BEARER_TOKEN", None)
|
||||
result = get_social_sentiment.invoke({"query": "polymarket"})
|
||||
|
||||
self.assertIsInstance(result, str)
|
||||
self.assertIn("Reddit", result)
|
||||
self.assertIn("Polymarket odds", result)
|
||||
|
||||
def test_error_in_twitter_graceful(self):
|
||||
mock_tweepy = MagicMock()
|
||||
mock_tweepy.Client.side_effect = Exception("auth error")
|
||||
|
||||
with patch.dict("sys.modules", {"tweepy": mock_tweepy}):
|
||||
with patch.dict("os.environ", {"TWITTER_BEARER_TOKEN": "fake-token"}):
|
||||
import os
|
||||
os.environ.pop("REDDIT_CLIENT_ID", None)
|
||||
result = get_social_sentiment.invoke({"query": "test"})
|
||||
|
||||
self.assertIsInstance(result, str)
|
||||
# Should not raise; should contain Twitter section mention
|
||||
self.assertIn("Twitter", result)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# search_markets
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestSearchMarkets(unittest.TestCase):
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_returns_string(self, mock_api):
|
||||
mock_api.return_value = [
|
||||
{"title": "US Election 2026", "volume": "1000000", "endDate": "2026-11-03", "active": True},
|
||||
{"title": "Super Bowl 2026", "volume": "500000", "endDate": "2026-02-07", "active": True},
|
||||
]
|
||||
result = search_markets.invoke({"min_volume": 10000})
|
||||
self.assertIsInstance(result, str)
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_contains_market_titles(self, mock_api):
|
||||
mock_api.return_value = [
|
||||
{"title": "US Election 2026", "volume": "1000000", "endDate": "2026-11-03", "active": True},
|
||||
]
|
||||
result = search_markets.invoke({"min_volume": 10000})
|
||||
self.assertIn("US Election 2026", result)
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_volume_filter(self, mock_api):
|
||||
mock_api.return_value = [
|
||||
{"title": "Big Market", "volume": "1000000", "endDate": "2026-11-03", "active": True},
|
||||
{"title": "Small Market", "volume": "500", "endDate": "2026-11-03", "active": True},
|
||||
]
|
||||
result = search_markets.invoke({"min_volume": 10000})
|
||||
self.assertIn("Big Market", result)
|
||||
self.assertNotIn("Small Market", result)
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_empty_results(self, mock_api):
|
||||
mock_api.return_value = []
|
||||
result = search_markets.invoke({"min_volume": 10000})
|
||||
self.assertIsInstance(result, str)
|
||||
self.assertIn("No markets matched", result)
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_error_handling(self, mock_api):
|
||||
mock_api.side_effect = Exception("server down")
|
||||
result = search_markets.invoke({"min_volume": 10000})
|
||||
self.assertIsInstance(result, str)
|
||||
self.assertIn("Error", result)
|
||||
|
||||
@patch("tradingagents.agents.utils.polymarket_tools._api_get")
|
||||
def test_category_filter_passed_to_api(self, mock_api):
|
||||
mock_api.return_value = [
|
||||
{"title": "Politics Market", "volume": "500000", "endDate": "2026-12-01", "active": True},
|
||||
]
|
||||
result = search_markets.invoke({"min_volume": 1000, "category": "politics"})
|
||||
self.assertIsInstance(result, str)
|
||||
# Verify category was passed in params
|
||||
call_kwargs = mock_api.call_args
|
||||
params = call_kwargs[1].get("params", call_kwargs[0][1] if len(call_kwargs[0]) > 1 else {})
|
||||
self.assertEqual(params.get("tag"), "politics")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
|
@ -0,0 +1,700 @@
|
|||
"""
|
||||
Polymarket API tool functions for the TradingAgents prediction market framework.
|
||||
|
||||
API base URLs:
|
||||
- Gamma: https://gamma-api.polymarket.com
|
||||
- CLOB: https://clob.polymarket.com
|
||||
- Data: https://data-api.polymarket.com
|
||||
"""
|
||||
|
||||
import os
|
||||
import json
|
||||
import time
|
||||
from typing import Optional
|
||||
import requests
|
||||
from langchain_core.tools import tool
|
||||
|
||||
GAMMA_BASE = "https://gamma-api.polymarket.com"
|
||||
CLOB_BASE = "https://clob.polymarket.com"
|
||||
DATA_BASE = "https://data-api.polymarket.com"
|
||||
|
||||
_MAX_RETRIES = 3
|
||||
_TIMEOUT = 30
|
||||
|
||||
|
||||
def _api_get(url: str, params: Optional[dict] = None) -> dict:
|
||||
"""
|
||||
Retry wrapper for GET requests with exponential backoff.
|
||||
Raises requests.HTTPError on non-2xx after max retries.
|
||||
"""
|
||||
last_exc: Exception = RuntimeError("_api_get failed with no attempts")
|
||||
for attempt in range(_MAX_RETRIES):
|
||||
try:
|
||||
resp = requests.get(url, params=params, timeout=_TIMEOUT)
|
||||
resp.raise_for_status()
|
||||
return resp.json()
|
||||
except (requests.RequestException, ValueError) as exc:
|
||||
last_exc = exc
|
||||
if attempt < _MAX_RETRIES - 1:
|
||||
time.sleep(2 ** attempt)
|
||||
raise last_exc
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tool 1: get_market_data
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@tool
|
||||
def get_market_data(event_id: str) -> str:
|
||||
"""
|
||||
Fetch market data for a Polymarket event.
|
||||
|
||||
Queries the Gamma API for the given event_id and returns a Markdown report
|
||||
including event metadata, outcome prices, volume, and spread.
|
||||
|
||||
Args:
|
||||
event_id: The Polymarket event ID (slug or numeric ID).
|
||||
|
||||
Returns:
|
||||
A formatted Markdown string with event metadata and market prices.
|
||||
"""
|
||||
try:
|
||||
data = _api_get(f"{GAMMA_BASE}/events/{event_id}")
|
||||
except Exception as exc:
|
||||
return f"## Error fetching market data\n\nFailed to retrieve data for event `{event_id}`:\n{exc}"
|
||||
|
||||
title = data.get("title", "N/A")
|
||||
slug = data.get("slug", "N/A")
|
||||
volume = data.get("volume", "N/A")
|
||||
liquidity = data.get("liquidity", "N/A")
|
||||
start_date = data.get("startDate", "N/A")
|
||||
end_date = data.get("endDate", "N/A")
|
||||
status = data.get("active", "N/A")
|
||||
|
||||
lines = [
|
||||
f"## Market Data: {title}",
|
||||
"",
|
||||
f"- **Event ID**: {event_id}",
|
||||
f"- **Slug**: {slug}",
|
||||
f"- **Status**: {'Active' if status else 'Closed'}",
|
||||
f"- **Start Date**: {start_date}",
|
||||
f"- **End Date**: {end_date}",
|
||||
f"- **Volume**: {volume}",
|
||||
f"- **Liquidity**: {liquidity}",
|
||||
"",
|
||||
"### Markets",
|
||||
]
|
||||
|
||||
markets = data.get("markets", [])
|
||||
for mkt in markets:
|
||||
q = mkt.get("question", "N/A")
|
||||
raw_prices = mkt.get("outcomePrices", "[]")
|
||||
raw_outcomes = mkt.get("outcomes", "[]")
|
||||
try:
|
||||
prices = json.loads(raw_prices) if isinstance(raw_prices, str) else raw_prices
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
prices = []
|
||||
try:
|
||||
outcomes = json.loads(raw_outcomes) if isinstance(raw_outcomes, str) else raw_outcomes
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
outcomes = []
|
||||
|
||||
spread = mkt.get("spread", "N/A")
|
||||
mkt_volume = mkt.get("volume", "N/A")
|
||||
|
||||
lines.append(f"\n**{q}**")
|
||||
for i, outcome in enumerate(outcomes):
|
||||
price = prices[i] if i < len(prices) else "N/A"
|
||||
lines.append(f" - {outcome}: {price}")
|
||||
lines.append(f" - Spread: {spread} | Volume: {mkt_volume}")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tool 2: get_price_history
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@tool
|
||||
def get_price_history(token_id: str, interval: str = "1w") -> str:
|
||||
"""
|
||||
Fetch price history for a Polymarket token.
|
||||
|
||||
Queries the CLOB /prices-history endpoint and returns a time-series table
|
||||
plus summary statistics (min, max, mean, latest price).
|
||||
|
||||
Args:
|
||||
token_id: The CLOB token ID for the market outcome.
|
||||
interval: Time interval string (e.g., "1d", "1w", "1m"). Defaults to "1w".
|
||||
|
||||
Returns:
|
||||
A formatted Markdown string with price history and stats.
|
||||
"""
|
||||
try:
|
||||
data = _api_get(f"{CLOB_BASE}/prices-history", params={"market": token_id, "interval": interval, "fidelity": 60})
|
||||
except Exception as exc:
|
||||
return f"## Error fetching price history\n\nFailed for token `{token_id}`:\n{exc}"
|
||||
|
||||
history = data.get("history", [])
|
||||
if not history:
|
||||
return f"## Price History: {token_id}\n\nNo price history data available."
|
||||
|
||||
prices = [float(h.get("p", 0)) for h in history if h.get("p") is not None]
|
||||
timestamps = [h.get("t", "") for h in history]
|
||||
|
||||
min_p = min(prices) if prices else 0
|
||||
max_p = max(prices) if prices else 0
|
||||
mean_p = sum(prices) / len(prices) if prices else 0
|
||||
latest_p = prices[-1] if prices else 0
|
||||
|
||||
lines = [
|
||||
f"## Price History: Token {token_id}",
|
||||
f"**Interval**: {interval}",
|
||||
"",
|
||||
"### Summary Statistics",
|
||||
f"- **Latest Price**: {latest_p:.4f}",
|
||||
f"- **Min Price**: {min_p:.4f}",
|
||||
f"- **Max Price**: {max_p:.4f}",
|
||||
f"- **Mean Price**: {mean_p:.4f}",
|
||||
f"- **Data Points**: {len(prices)}",
|
||||
"",
|
||||
"### Price Series (last 20 entries)",
|
||||
"| Timestamp | Price |",
|
||||
"|-----------|-------|",
|
||||
]
|
||||
|
||||
recent = list(zip(timestamps, prices))[-20:]
|
||||
for ts, p in recent:
|
||||
lines.append(f"| {ts} | {p:.4f} |")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tool 3: get_event_details
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@tool
|
||||
def get_event_details(event_id: str) -> str:
|
||||
"""
|
||||
Fetch detailed information about a Polymarket event.
|
||||
|
||||
Queries the Gamma API and returns description, resolution criteria,
|
||||
resolution deadline, and associated markets.
|
||||
|
||||
Args:
|
||||
event_id: The Polymarket event ID.
|
||||
|
||||
Returns:
|
||||
A formatted Markdown string with event details.
|
||||
"""
|
||||
try:
|
||||
data = _api_get(f"{GAMMA_BASE}/events/{event_id}")
|
||||
except Exception as exc:
|
||||
return f"## Error fetching event details\n\nFailed for event `{event_id}`:\n{exc}"
|
||||
|
||||
title = data.get("title", "N/A")
|
||||
description = data.get("description", "No description available.")
|
||||
resolution_source = data.get("resolutionSource", "N/A")
|
||||
end_date = data.get("endDate", "N/A")
|
||||
tags = data.get("tags", [])
|
||||
tag_names = [t.get("label", "") for t in tags] if isinstance(tags, list) else []
|
||||
|
||||
lines = [
|
||||
f"## Event Details: {title}",
|
||||
"",
|
||||
f"**Event ID**: {event_id}",
|
||||
f"**Resolution Deadline**: {end_date}",
|
||||
f"**Resolution Source**: {resolution_source}",
|
||||
f"**Tags**: {', '.join(tag_names) if tag_names else 'None'}",
|
||||
"",
|
||||
"### Description",
|
||||
description,
|
||||
"",
|
||||
"### Associated Markets",
|
||||
]
|
||||
|
||||
markets = data.get("markets", [])
|
||||
for mkt in markets:
|
||||
q = mkt.get("question", "N/A")
|
||||
cond_id = mkt.get("conditionId", "N/A")
|
||||
status = "Active" if mkt.get("active") else "Closed"
|
||||
end = mkt.get("endDate", "N/A")
|
||||
resolution_criteria = mkt.get("description", "")
|
||||
lines.append(f"\n#### {q}")
|
||||
lines.append(f"- **Condition ID**: {cond_id}")
|
||||
lines.append(f"- **Status**: {status}")
|
||||
lines.append(f"- **End Date**: {end}")
|
||||
if resolution_criteria:
|
||||
lines.append(f"- **Resolution Criteria**: {resolution_criteria[:300]}")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tool 4: get_orderbook
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@tool
|
||||
def get_orderbook(token_id: str) -> str:
|
||||
"""
|
||||
Fetch the current order book for a Polymarket token.
|
||||
|
||||
Queries the CLOB /book endpoint and returns bid/ask depth tables.
|
||||
|
||||
Args:
|
||||
token_id: The CLOB token ID for the market outcome.
|
||||
|
||||
Returns:
|
||||
A formatted Markdown string with bid and ask depth tables.
|
||||
"""
|
||||
try:
|
||||
data = _api_get(f"{CLOB_BASE}/book", params={"token_id": token_id})
|
||||
except Exception as exc:
|
||||
return f"## Error fetching order book\n\nFailed for token `{token_id}`:\n{exc}"
|
||||
|
||||
bids = data.get("bids", [])
|
||||
asks = data.get("asks", [])
|
||||
market = data.get("market", token_id)
|
||||
asset_id = data.get("asset_id", token_id)
|
||||
|
||||
lines = [
|
||||
f"## Order Book: {asset_id}",
|
||||
f"**Market**: {market}",
|
||||
"",
|
||||
"### Bids (Buy Orders)",
|
||||
"| Price | Size |",
|
||||
"|-------|------|",
|
||||
]
|
||||
for bid in bids[:10]:
|
||||
p = bid.get("price", "N/A")
|
||||
s = bid.get("size", "N/A")
|
||||
lines.append(f"| {p} | {s} |")
|
||||
|
||||
lines += [
|
||||
"",
|
||||
"### Asks (Sell Orders)",
|
||||
"| Price | Size |",
|
||||
"|-------|------|",
|
||||
]
|
||||
for ask in asks[:10]:
|
||||
p = ask.get("price", "N/A")
|
||||
s = ask.get("size", "N/A")
|
||||
lines.append(f"| {p} | {s} |")
|
||||
|
||||
if bids and asks:
|
||||
best_bid = float(bids[0].get("price", 0))
|
||||
best_ask = float(asks[0].get("price", 0))
|
||||
spread = best_ask - best_bid
|
||||
lines += ["", f"**Best Bid**: {best_bid:.4f} | **Best Ask**: {best_ask:.4f} | **Spread**: {spread:.4f}"]
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tool 5: get_event_news
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@tool
|
||||
def get_event_news(query: str, api_key: Optional[str] = None) -> str:
|
||||
"""
|
||||
Search for news articles related to a Polymarket event using Tavily.
|
||||
|
||||
Args:
|
||||
query: Search query string related to the prediction market event.
|
||||
api_key: Tavily API key. Falls back to TAVILY_API_KEY environment variable.
|
||||
|
||||
Returns:
|
||||
A formatted Markdown string with relevant news articles.
|
||||
"""
|
||||
key = api_key or os.getenv("TAVILY_API_KEY")
|
||||
if not key:
|
||||
return "## Event News\n\nNo Tavily API key provided. Set TAVILY_API_KEY environment variable."
|
||||
|
||||
try:
|
||||
from tavily import TavilyClient
|
||||
client = TavilyClient(api_key=key)
|
||||
response = client.search(query=query, max_results=5)
|
||||
except ImportError:
|
||||
return "## Event News\n\nTavily package is not installed. Run: pip install tavily-python"
|
||||
except Exception as exc:
|
||||
return f"## Event News\n\nSearch failed for query `{query}`:\n{exc}"
|
||||
|
||||
results = response.get("results", [])
|
||||
lines = [
|
||||
f"## Event News: {query}",
|
||||
f"**Query**: {query}",
|
||||
f"**Results**: {len(results)} articles",
|
||||
"",
|
||||
]
|
||||
for i, art in enumerate(results, 1):
|
||||
title = art.get("title", "N/A")
|
||||
url = art.get("url", "")
|
||||
content = art.get("content", "")[:400]
|
||||
published = art.get("published_date", "N/A")
|
||||
lines += [
|
||||
f"### {i}. {title}",
|
||||
f"**Published**: {published}",
|
||||
f"**URL**: {url}",
|
||||
f"{content}",
|
||||
"",
|
||||
]
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tool 6: get_global_news
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@tool
|
||||
def get_global_news(query: str = "prediction markets macro economic news", api_key: Optional[str] = None) -> str:
|
||||
"""
|
||||
Search for global macroeconomic and market news using Tavily.
|
||||
|
||||
Args:
|
||||
query: Search query for macro/global news. Defaults to a broad market query.
|
||||
api_key: Tavily API key. Falls back to TAVILY_API_KEY environment variable.
|
||||
|
||||
Returns:
|
||||
A formatted Markdown string with global news articles.
|
||||
"""
|
||||
key = api_key or os.getenv("TAVILY_API_KEY")
|
||||
if not key:
|
||||
return "## Global News\n\nNo Tavily API key provided. Set TAVILY_API_KEY environment variable."
|
||||
|
||||
try:
|
||||
from tavily import TavilyClient
|
||||
client = TavilyClient(api_key=key)
|
||||
response = client.search(query=query, max_results=5, search_depth="advanced")
|
||||
except ImportError:
|
||||
return "## Global News\n\nTavily package is not installed. Run: pip install tavily-python"
|
||||
except Exception as exc:
|
||||
return f"## Global News\n\nSearch failed:\n{exc}"
|
||||
|
||||
results = response.get("results", [])
|
||||
lines = [
|
||||
f"## Global News",
|
||||
f"**Query**: {query}",
|
||||
f"**Results**: {len(results)} articles",
|
||||
"",
|
||||
]
|
||||
for i, art in enumerate(results, 1):
|
||||
title = art.get("title", "N/A")
|
||||
url = art.get("url", "")
|
||||
content = art.get("content", "")[:400]
|
||||
published = art.get("published_date", "N/A")
|
||||
lines += [
|
||||
f"### {i}. {title}",
|
||||
f"**Published**: {published}",
|
||||
f"**URL**: {url}",
|
||||
f"{content}",
|
||||
"",
|
||||
]
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tool 7: get_whale_activity
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@tool
|
||||
def get_whale_activity(market_id: str) -> str:
|
||||
"""
|
||||
Fetch top holder (whale) activity for a Polymarket market.
|
||||
|
||||
Queries the Data API /holders endpoint and returns a table of top holders.
|
||||
|
||||
Args:
|
||||
market_id: The Polymarket market/condition ID.
|
||||
|
||||
Returns:
|
||||
A formatted Markdown string with top holder information.
|
||||
"""
|
||||
try:
|
||||
data = _api_get(f"{DATA_BASE}/holders", params={"market": market_id, "limit": 20})
|
||||
except Exception as exc:
|
||||
return f"## Error fetching whale activity\n\nFailed for market `{market_id}`:\n{exc}"
|
||||
|
||||
holders = data if isinstance(data, list) else data.get("holders", data.get("data", []))
|
||||
|
||||
lines = [
|
||||
f"## Whale Activity: {market_id}",
|
||||
"",
|
||||
"| Rank | Address | Position | Value |",
|
||||
"|------|---------|---------|-------|",
|
||||
]
|
||||
for i, holder in enumerate(holders[:20], 1):
|
||||
address = holder.get("proxyWallet", holder.get("address", "N/A"))
|
||||
position = holder.get("position", holder.get("amount", "N/A"))
|
||||
value = holder.get("value", holder.get("usdcValue", "N/A"))
|
||||
lines.append(f"| {i} | {address[:12]}... | {position} | {value} |")
|
||||
|
||||
if not holders:
|
||||
lines.append("| - | No holder data available | - | - |")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tool 8: get_market_stats
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@tool
|
||||
def get_market_stats(market_id: str) -> str:
|
||||
"""
|
||||
Fetch open interest and market statistics for a Polymarket market.
|
||||
|
||||
Queries the Data API /openInterest endpoint and returns OI and related stats.
|
||||
|
||||
Args:
|
||||
market_id: The Polymarket market/condition ID.
|
||||
|
||||
Returns:
|
||||
A formatted Markdown string with market statistics.
|
||||
"""
|
||||
try:
|
||||
data = _api_get(f"{DATA_BASE}/openInterest", params={"market": market_id})
|
||||
except Exception as exc:
|
||||
return f"## Error fetching market stats\n\nFailed for market `{market_id}`:\n{exc}"
|
||||
|
||||
oi = data.get("openInterest", data.get("open_interest", "N/A"))
|
||||
total_volume = data.get("totalVolume", data.get("volume", "N/A"))
|
||||
num_traders = data.get("numTraders", data.get("traders", "N/A"))
|
||||
liquidity = data.get("liquidity", "N/A")
|
||||
last_trade_price = data.get("lastTradePrice", "N/A")
|
||||
|
||||
lines = [
|
||||
f"## Market Statistics: {market_id}",
|
||||
"",
|
||||
f"- **Open Interest**: {oi}",
|
||||
f"- **Total Volume**: {total_volume}",
|
||||
f"- **Number of Traders**: {num_traders}",
|
||||
f"- **Liquidity**: {liquidity}",
|
||||
f"- **Last Trade Price**: {last_trade_price}",
|
||||
]
|
||||
|
||||
# Include any additional fields
|
||||
extra_fields = {k: v for k, v in data.items()
|
||||
if k not in ("openInterest", "open_interest", "totalVolume", "volume",
|
||||
"numTraders", "traders", "liquidity", "lastTradePrice")}
|
||||
if extra_fields:
|
||||
lines.append("\n### Additional Stats")
|
||||
for k, v in list(extra_fields.items())[:10]:
|
||||
lines.append(f"- **{k}**: {v}")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tool 9: get_leaderboard_signals
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@tool
|
||||
def get_leaderboard_signals(category: str = "OVERALL", time_period: str = "WEEK") -> str:
|
||||
"""
|
||||
Fetch leaderboard data from Polymarket to identify top traders and signals.
|
||||
|
||||
Queries the Data API /v1/leaderboard endpoint.
|
||||
|
||||
Args:
|
||||
category: Leaderboard category (e.g., "OVERALL", "POLITICS", "SPORTS"). Defaults to "OVERALL".
|
||||
time_period: Time period for the leaderboard (e.g., "WEEK", "MONTH", "ALL"). Defaults to "WEEK".
|
||||
|
||||
Returns:
|
||||
A formatted Markdown string with leaderboard rankings.
|
||||
"""
|
||||
try:
|
||||
data = _api_get(f"{DATA_BASE}/v1/leaderboard", params={"category": category, "timePeriod": time_period})
|
||||
except Exception as exc:
|
||||
return f"## Error fetching leaderboard\n\nFailed (category={category}, period={time_period}):\n{exc}"
|
||||
|
||||
traders = data if isinstance(data, list) else data.get("data", data.get("leaderboard", []))
|
||||
|
||||
lines = [
|
||||
f"## Leaderboard Signals",
|
||||
f"**Category**: {category} | **Period**: {time_period}",
|
||||
"",
|
||||
"| Rank | Name / Address | Profit & Loss | Volume |",
|
||||
"|------|---------------|--------------|--------|",
|
||||
]
|
||||
for i, trader in enumerate(traders[:20], 1):
|
||||
name = trader.get("name", trader.get("pseudonym", trader.get("proxyWallet", "N/A")))
|
||||
if len(str(name)) > 20:
|
||||
name = str(name)[:18] + "..."
|
||||
pnl = trader.get("pnl", trader.get("profit", "N/A"))
|
||||
volume = trader.get("volume", "N/A")
|
||||
lines.append(f"| {i} | {name} | {pnl} | {volume} |")
|
||||
|
||||
if not traders:
|
||||
lines.append("| - | No leaderboard data available | - | - |")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tool 10: get_social_sentiment
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@tool
|
||||
def get_social_sentiment(query: str) -> str:
|
||||
"""
|
||||
Fetch social media sentiment from Twitter and Reddit related to a query.
|
||||
|
||||
Gracefully skips Twitter/Reddit if API keys are not set in environment.
|
||||
Required environment variables:
|
||||
- Twitter: TWITTER_BEARER_TOKEN
|
||||
- Reddit: REDDIT_CLIENT_ID, REDDIT_CLIENT_SECRET, REDDIT_USER_AGENT
|
||||
|
||||
Args:
|
||||
query: The search query or topic for sentiment analysis.
|
||||
|
||||
Returns:
|
||||
A formatted Markdown string with social sentiment data.
|
||||
"""
|
||||
lines = [f"## Social Sentiment: {query}", ""]
|
||||
any_data = False
|
||||
|
||||
# --- Twitter ---
|
||||
bearer_token = os.getenv("TWITTER_BEARER_TOKEN")
|
||||
if bearer_token:
|
||||
try:
|
||||
import tweepy
|
||||
client = tweepy.Client(bearer_token=bearer_token)
|
||||
response = client.search_recent_tweets(
|
||||
query=f"{query} -is:retweet lang:en",
|
||||
max_results=10,
|
||||
tweet_fields=["created_at", "public_metrics"],
|
||||
)
|
||||
tweets = response.data or []
|
||||
lines += [
|
||||
"### Twitter",
|
||||
f"**Recent tweets** ({len(tweets)} found):",
|
||||
"",
|
||||
]
|
||||
for tweet in tweets:
|
||||
text = tweet.text[:200].replace("\n", " ")
|
||||
metrics = getattr(tweet, "public_metrics", {}) or {}
|
||||
likes = metrics.get("like_count", 0)
|
||||
rts = metrics.get("retweet_count", 0)
|
||||
lines.append(f"- {text} _(likes: {likes}, retweets: {rts})_")
|
||||
any_data = True
|
||||
except ImportError:
|
||||
lines.append("_Twitter: tweepy not installed (pip install tweepy)_")
|
||||
except Exception as exc:
|
||||
lines.append(f"_Twitter: Error — {exc}_")
|
||||
else:
|
||||
lines.append("_Twitter: TWITTER_BEARER_TOKEN not set — skipping._")
|
||||
|
||||
lines.append("")
|
||||
|
||||
# --- Reddit ---
|
||||
reddit_id = os.getenv("REDDIT_CLIENT_ID")
|
||||
reddit_secret = os.getenv("REDDIT_CLIENT_SECRET")
|
||||
reddit_agent = os.getenv("REDDIT_USER_AGENT", "polymarket-agent/1.0")
|
||||
if reddit_id and reddit_secret:
|
||||
try:
|
||||
import praw
|
||||
reddit = praw.Reddit(
|
||||
client_id=reddit_id,
|
||||
client_secret=reddit_secret,
|
||||
user_agent=reddit_agent,
|
||||
)
|
||||
subreddits = "Polymarket+PredictionMarkets+politics+finance"
|
||||
results = list(reddit.subreddit(subreddits).search(query, limit=10, sort="new"))
|
||||
lines += [
|
||||
"### Reddit",
|
||||
f"**Recent posts** ({len(results)} found):",
|
||||
"",
|
||||
]
|
||||
for post in results:
|
||||
title = post.title[:150].replace("\n", " ")
|
||||
score = post.score
|
||||
sub = post.subreddit.display_name
|
||||
lines.append(f"- [{sub}] {title} _(score: {score})_")
|
||||
any_data = True
|
||||
except ImportError:
|
||||
lines.append("_Reddit: praw not installed (pip install praw)_")
|
||||
except Exception as exc:
|
||||
lines.append(f"_Reddit: Error — {exc}_")
|
||||
else:
|
||||
lines.append("_Reddit: REDDIT_CLIENT_ID / REDDIT_CLIENT_SECRET not set — skipping._")
|
||||
|
||||
if not any_data:
|
||||
lines.append("\n_No social sentiment data available. Configure API keys._")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tool 11: search_markets
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@tool
|
||||
def search_markets(
|
||||
min_volume: int = 10000,
|
||||
category: str = "",
|
||||
status: str = "active",
|
||||
limit: int = 20,
|
||||
) -> str:
|
||||
"""
|
||||
Search for Polymarket events/markets with optional filters.
|
||||
|
||||
Queries the Gamma API /events endpoint with filters for volume, category,
|
||||
and market status.
|
||||
|
||||
Args:
|
||||
min_volume: Minimum trading volume filter (default: 10000).
|
||||
category: Category tag filter, e.g., "politics", "sports" (optional).
|
||||
status: Market status filter: "active" or "closed" (default: "active").
|
||||
limit: Maximum number of results to return (default: 20).
|
||||
|
||||
Returns:
|
||||
A formatted Markdown string listing matching markets.
|
||||
"""
|
||||
params = {
|
||||
"limit": min(limit, 100),
|
||||
"active": (status.lower() == "active"),
|
||||
"closed": (status.lower() == "closed"),
|
||||
}
|
||||
if category:
|
||||
params["tag"] = category
|
||||
|
||||
try:
|
||||
data = _api_get(f"{GAMMA_BASE}/events", params=params)
|
||||
except Exception as exc:
|
||||
return f"## Error searching markets\n\nFailed:\n{exc}"
|
||||
|
||||
events = data if isinstance(data, list) else data.get("events", data.get("data", []))
|
||||
|
||||
# Filter by min_volume
|
||||
filtered = []
|
||||
for ev in events:
|
||||
try:
|
||||
vol = float(ev.get("volume", 0) or 0)
|
||||
except (ValueError, TypeError):
|
||||
vol = 0
|
||||
if vol >= min_volume:
|
||||
filtered.append(ev)
|
||||
|
||||
lines = [
|
||||
"## Market Search Results",
|
||||
f"**Filters**: min_volume={min_volume}, category='{category}', status={status}",
|
||||
f"**Found**: {len(filtered)} markets (showing up to {limit})",
|
||||
"",
|
||||
"| Title | Volume | End Date | Status |",
|
||||
"|-------|--------|---------|--------|",
|
||||
]
|
||||
|
||||
for ev in filtered[:limit]:
|
||||
title = str(ev.get("title", "N/A"))[:50]
|
||||
vol = ev.get("volume", "N/A")
|
||||
end = ev.get("endDate", "N/A")
|
||||
active = "Active" if ev.get("active") else "Closed"
|
||||
lines.append(f"| {title} | {vol} | {end} | {active} |")
|
||||
|
||||
if not filtered:
|
||||
lines.append("| No markets matched the given filters | - | - | - |")
|
||||
|
||||
return "\n".join(lines)
|
||||
Loading…
Reference in New Issue