1
0
mirror of https://gitlab.com/simple-stock-bots/simple-stock-bot.git synced 2026-06-03 21:00:26 +00:00

Compare commits

..

8 Commits

Author SHA1 Message Date
Anson 5baf7e42c0 update reqs 2024-05-12 21:48:24 -06:00
Anson 1bde1fbf19 just need ci to pass so I can update the site and shut things down 2024-05-12 21:42:08 -06:00
Anson d60b14c9de update site with archive notice 2024-05-13 03:40:32 +00:00
Anson ec516db22f formatting 2023-11-30 21:53:06 -07:00
Anson 1fe7fe8c9c formatting 2023-11-30 21:50:25 -07:00
Anson 0c71193194 weirdo bug 2023-11-30 21:50:17 -07:00
Anson d3b75984a6 full on ruff 2023-11-30 21:49:56 -07:00
Anson b3769b24d8 dont need black anymore 2023-11-30 21:06:35 -07:00
20 changed files with 123 additions and 566 deletions
-17
View File
@@ -1,17 +0,0 @@
# Required Environment Variables for Simple Stock Bot
# Telegram Bot Token (required for Telegram functionality)
# Get this from @BotFather on Telegram
TELEGRAM=your_telegram_bot_token_here
# Discord Bot Token (required for Discord functionality)
# Get this from Discord Developer Portal
DISCORD=your_discord_bot_token_here
# MarketData.app API Token (required for stock market data)
# Get this from https://www.marketdata.app/
MARKETDATA=your_marketdata_api_token_here
# Stripe Token (optional - for accepting donations)
# Get this from Stripe Dashboard
STRIPE=your_stripe_token_here
+6 -8
View File
@@ -3,14 +3,12 @@ stages:
- build_site - build_site
- deploy - deploy
ruff: # ruff:
stage: lint # stage: lint
image: python:3.12-slim # image: python:3.11-slim
script: # script:
- pip3 install ruff==0.15.2 # - pip3 install ruff
- ruff check . --output-format gitlab # - ruff . --output-format gitlab; ruff format . --diff
- ruff format . --check
allow_failure: false
# prettier: # prettier:
# stage: lint # stage: lint
+15 -9
View File
@@ -74,7 +74,7 @@ class MarketData:
resp = r.get(url, params=params, timeout=timeout, headers=headers) resp = r.get(url, params=params, timeout=timeout, headers=headers)
logging.debug(resp.headers.items()) logging.error(resp.headers.items())
# Make sure API returned a proper status code # Make sure API returned a proper status code
try: try:
@@ -133,21 +133,27 @@ class MarketData:
self.charts = {} self.charts = {}
def status(self) -> str: def status(self) -> str:
"""Check MarketData.app API status by making a test request.""" # TODO: At the moment this API is poorly documented, this function likely needs to be revisited later.
try: try:
# Test the API with a simple request
status = r.get( status = r.get(
"https://api.marketdata.app/v1/stocks/quotes/AAPL/", "https://stats.uptimerobot.com/api/getMonitorList/6Kv3zIow0A",
timeout=5, timeout=5,
) )
status.raise_for_status() status.raise_for_status()
return f"MarketData.app API is responding OK with status {status.status_code} in {status.elapsed.total_seconds():.2f} seconds."
except r.HTTPError: except r.HTTPError:
return f"MarketData.app API returned an HTTP error code {status.status_code} in {status.elapsed.total_seconds():.2f} seconds." return f"API returned an HTTP error code {status.status_code} in {status.elapsed.total_seconds()} seconds."
except r.Timeout: except r.Timeout:
return "MarketData.app API timed out before it could respond. This is likely due to a surge in usage or a complete outage." return "API timed out before it was able to give status. This is likely due to a surge in usage or a complete outage."
except Exception as e:
return f"MarketData.app API check failed: {str(e)}" statusJSON = status.json()
if statusJSON["status"] == "ok":
return (
f"CoinGecko API responded that it was OK with a {status.status_code} in {status.elapsed.total_seconds()} seconds."
)
else:
return f"MarketData.app is currently reporting the following status: {statusJSON['status']}"
def price_reply(self, symbol: Stock) -> str: def price_reply(self, symbol: Stock) -> str:
"""Returns price movement of Stock for the last market day, or after hours. """Returns price movement of Stock for the last market day, or after hours.
+12 -21
View File
@@ -31,24 +31,15 @@ class cg_Crypto:
# This results in a rate limit of 15 requests per minute for each bot. # This results in a rate limit of 15 requests per minute for each bot.
# Given this, the rate limit effectively becomes 1 request every 4 seconds for each bot. # Given this, the rate limit effectively becomes 1 request every 4 seconds for each bot.
@rate_limited(0.25) @rate_limited(0.25)
def get(self, endpoint, params: dict = {}, timeout=10, retry_count=0, max_retries=3) -> dict: def get(self, endpoint, params: dict = {}, timeout=10) -> dict:
url = "https://api.coingecko.com/api/v3" + endpoint url = "https://api.coingecko.com/api/v3" + endpoint
resp = r.get(url, params=params, timeout=timeout) resp = r.get(url, params=params, timeout=timeout)
# Make sure API returned a proper status code # Make sure API returned a proper status code
if resp.status_code == 429: if resp.status_code == 429:
if retry_count >= max_retries: log.warning(f"CoinGecko returned 429 - Too Many Requests for endpoint: {endpoint}. Sleeping and trying again.")
log.error(f"CoinGecko 429 retry limit ({max_retries}) exceeded for endpoint: {endpoint}") time.sleep(10)
return {} return self.get(endpoint=endpoint, params=params, timeout=timeout)
backoff_time = (2**retry_count) * 10 # Exponential backoff: 10s, 20s, 40s
log.warning(
f"CoinGecko returned 429 - Too Many Requests for endpoint: {endpoint}. Retry {retry_count + 1}/{max_retries} after {backoff_time}s."
)
time.sleep(backoff_time)
return self.get(
endpoint=endpoint, params=params, timeout=timeout, retry_count=retry_count + 1, max_retries=max_retries
)
try: try:
resp.raise_for_status() resp.raise_for_status()
@@ -223,14 +214,14 @@ class cg_Crypto:
}, },
): ):
return f""" return f"""
[{data["name"]}]({data["links"]["homepage"][0]}) Statistics: [{data['name']}]({data['links']['homepage'][0]}) Statistics:
Market Cap: ${data["market_data"]["market_cap"][self.vs_currency]:,} Market Cap: ${data['market_data']['market_cap'][self.vs_currency]:,}
Market Cap Ranking: {data.get("market_cap_rank", "Not Available")} Market Cap Ranking: {data.get('market_cap_rank',"Not Available")}
CoinGecko Scores: CoinGecko Scores:
Overall: {data.get("coingecko_score", "Not Available")} Overall: {data.get('coingecko_score','Not Available')}
Development: {data.get("developer_score", "Not Available")} Development: {data.get('developer_score','Not Available')}
Community: {data.get("community_score", "Not Available")} Community: {data.get('community_score','Not Available')}
Public Interest: {data.get("public_interest_score", "Not Available")} Public Interest: {data.get('public_interest_score','Not Available')}
""" """
else: else:
return f"{symbol.symbol} returned an error." return f"{symbol.symbol} returned an error."
@@ -270,7 +261,7 @@ class cg_Crypto:
message = ( message = (
f"The current price of {coin.name} is $**{price:,}** and" f"The current price of {coin.name} is $**{price:,}** and"
+ f" its market cap is $**{cap:,.2f}** {self.vs_currency.upper()}" + " its market cap is $**{cap:,.2f}** {self.vs_currency.upper()}"
) )
else: else:
+7 -7
View File
@@ -1,8 +1,8 @@
cachetools==7.0.1 cachetools==5.3.1
humanize==4.15.0 humanize==4.8.0
markdownify==1.2.2 markdownify==0.11.6
mplfinance==0.12.10a7 mplfinance==0.12.10b0
pandas==3.0.1 pandas==2.1.1
requests==2.32.5 requests==2.31.0
rush==2021.4.0 rush==2021.4.0
schedule==1.2.2 schedule==1.2.1
+2 -1
View File
@@ -1,4 +1,5 @@
"""Function that routes symbols to the correct API provider.""" """Function that routes symbols to the correct API provider.
"""
import datetime import datetime
import logging import logging
+1 -3
View File
@@ -7,6 +7,4 @@ pylama==8.4.1
mypy==1.5.1 mypy==1.5.1
types-cachetools==5.3.0.6 types-cachetools==5.3.0.6
types-pytz==2023.3.1.1 types-pytz==2023.3.1.1
ruff==0.15.2 ruff==0.1.6
pytest==9.0.2
pytest-asyncio==0.25.3
+2 -1
View File
@@ -1,4 +1,5 @@
"""Functions and Info specific to the discord Bot""" """Functions and Info specific to the discord Bot
"""
import re import re
+2 -2
View File
@@ -1,4 +1,4 @@
FROM python:3.12-bookworm AS builder FROM python:3.11-buster AS builder
COPY discord/discord-reqs.txt . COPY discord/discord-reqs.txt .
@@ -7,7 +7,7 @@ COPY common/requirements.txt .
RUN pip install --user -r discord-reqs.txt RUN pip install --user -r discord-reqs.txt
FROM python:3.12-slim FROM python:3.11-slim
ENV MPLBACKEND=Agg ENV MPLBACKEND=Agg
+4 -2
View File
@@ -217,7 +217,9 @@ async def handle_options(message, symbols):
embed = nextcord.Embed(title=options_data["Option Symbol"], description=options_data["Underlying"], color=0x3498DB) embed = nextcord.Embed(title=options_data["Option Symbol"], description=options_data["Underlying"], color=0x3498DB)
# Key details # Key details
details = f"Expiration: {options_data['Expiration']}\nSide: {options_data['side']}\nStrike: {options_data['strike']}" details = (
f"Expiration: {options_data['Expiration']}\n" f"Side: {options_data['side']}\n" f"Strike: {options_data['strike']}"
)
embed.add_field(name="Details", value=details, inline=False) embed.add_field(name="Details", value=details, inline=False)
# Pricing info # Pricing info
@@ -230,7 +232,7 @@ async def handle_options(message, symbols):
embed.add_field(name="Pricing", value=pricing_info, inline=False) embed.add_field(name="Pricing", value=pricing_info, inline=False)
# Volume and open interest # Volume and open interest
volume_info = f"Open Interest: {options_data['Open Interest']}\nVolume: {options_data['Volume']}" volume_info = f"Open Interest: {options_data['Open Interest']}\n" f"Volume: {options_data['Volume']}"
embed.add_field(name="Activity", value=volume_info, inline=False) embed.add_field(name="Activity", value=volume_info, inline=False)
# Greeks # Greeks
+1 -1
View File
@@ -9,6 +9,6 @@ description: >
## What is Simple Stock Bot? ## What is Simple Stock Bot?
For full details see my full blog post: [https://notes.ansonbiggs.com/simple-stock-bot-end-of-life/](https://notes.ansonbiggs.com/simple-stock-bot-end-of-life/) For full details see my full blog post: https://notes.ansonbiggs.com/simple-stock-bot-end-of-life/
I am shutting down Simple Stock Bot, a popular Telegram and Discord bot that provided live stock and cryptocurrency market data. The bot was created in 2018 after my group chat lost access to the Google Allo, which had integrated market data. The bot grew to 15,000 monthly active users, but maintaining the service has become challenging due to increasing data costs and changes to the Discord API. I really appreciate the community and everyone that has donated along the way, this has been a seriously great ride. The bot will soon no longer function on Telegram and Discord. I am shutting down Simple Stock Bot, a popular Telegram and Discord bot that provided live stock and cryptocurrency market data. The bot was created in 2018 after my group chat lost access to the Google Allo, which had integrated market data. The bot grew to 15,000 monthly active users, but maintaining the service has become challenging due to increasing data costs and changes to the Discord API. I really appreciate the community and everyone that has donated along the way, this has been a seriously great ride. The bot will soon no longer function on Telegram and Discord.
+2 -2
View File
@@ -1,4 +1,4 @@
FROM python:3.12-bookworm AS builder FROM python:3.11-buster AS builder
COPY telegram/telegram-reqs.txt . COPY telegram/telegram-reqs.txt .
@@ -7,7 +7,7 @@ COPY common/requirements.txt .
RUN pip install --user -r telegram-reqs.txt RUN pip install --user -r telegram-reqs.txt
FROM python:3.12-slim FROM python:3.11-slim
ENV MPLBACKEND=Agg ENV MPLBACKEND=Agg
+2 -1
View File
@@ -1,4 +1,5 @@
"""Functions and Info specific to the Telegram Bot""" """Functions and Info specific to the Telegram Bot
"""
import re import re
+1 -1
View File
@@ -232,7 +232,7 @@ def generate_options_reply(options_data: dict):
message_text += pricing_info message_text += pricing_info
# Volume and open interest # Volume and open interest
volume_info = f"*Open Interest:* `{options_data['Open Interest']}`\n*Volume:* `{options_data['Volume']}`\n\n" volume_info = f"*Open Interest:* `{options_data['Open Interest']}`\n" f"*Volume:* `{options_data['Volume']}`\n\n"
message_text += volume_info message_text += volume_info
# Greeks # Greeks
+1 -1
View File
@@ -1,2 +1,2 @@
python-telegram-bot==21.11.1 python-telegram-bot==20.6
-r requirements.txt -r requirements.txt
View File
-15
View File
@@ -1,15 +0,0 @@
"""Pytest configuration and fixtures."""
import sys
import os
from pathlib import Path
# Add the project root to Python path so we can import modules
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root))
# Set environment variables for testing
os.environ.setdefault("TELEGRAM", "test_token")
os.environ.setdefault("DISCORD", "test_token")
os.environ.setdefault("MARKETDATA", "test_token")
os.environ.setdefault("STRIPE", "test_token")
-132
View File
@@ -1,132 +0,0 @@
"""Test the rate_limited decorator functionality."""
import time
from unittest.mock import patch
from common.utilities import rate_limited
class TestRateLimited:
"""Test rate_limited decorator functionality."""
def test_rate_limited_decorator(self):
"""Test that rate limiting works correctly."""
call_times = []
@rate_limited(2.0) # 2 calls per second max
def test_function():
call_times.append(time.time())
return len(call_times)
# First call should execute immediately
result1 = test_function()
assert result1 == 1
assert len(call_times) == 1
# Second call should be delayed
start_time = time.time()
result2 = test_function()
end_time = time.time()
assert result2 == 2
assert len(call_times) == 2
# Should have waited at least close to 0.5 seconds (1/2 calls per second)
elapsed = end_time - start_time
assert elapsed >= 0.4 # Allow some tolerance for timing
def test_rate_limited_with_args(self):
"""Test rate limited function with arguments."""
@rate_limited(1.0) # 1 call per second
def test_function(x, y=1):
return x + y
# Test that arguments are passed correctly
result1 = test_function(5, y=10)
assert result1 == 15
# Test positional and keyword args
start_time = time.time()
result2 = test_function(3, 7)
end_time = time.time()
assert result2 == 10
# Should have waited about 1 second
elapsed = end_time - start_time
assert elapsed >= 0.9
def test_rate_limited_logging(self):
"""Test that rate limiting logs waiting messages."""
@rate_limited(10.0) # 10 calls per second
def test_function():
return "called"
with patch("common.utilities.log.info") as mock_log:
# First call should not log
test_function()
mock_log.assert_not_called()
# Second call should log the wait
test_function()
mock_log.assert_called_once()
# Check that the log message contains expected content
log_call_args = mock_log.call_args[0][0]
assert "Rate limit exceeded" in log_call_args
assert "Waiting for" in log_call_args
def test_rate_limited_different_rates(self):
"""Test different rate limiting values."""
@rate_limited(4.0) # 4 calls per second = 0.25 second intervals
def fast_function():
return time.time()
@rate_limited(0.5) # 0.5 calls per second = 2 second intervals
def slow_function():
return time.time()
# Test fast function (should wait ~0.25 seconds)
start = time.time()
fast_function()
fast_function()
fast_elapsed = time.time() - start
assert 0.2 <= fast_elapsed <= 0.4
# Test slow function (should wait ~2 seconds)
start = time.time()
slow_function()
slow_function()
slow_elapsed = time.time() - start
assert slow_elapsed >= 1.9
def test_multiple_instances(self):
"""Test that different decorated functions maintain separate rate limits."""
@rate_limited(2.0)
def function_a():
return "A"
@rate_limited(2.0)
def function_b():
return "B"
# Both functions should be able to call immediately initially
result_a = function_a()
result_b = function_b()
assert result_a == "A"
assert result_b == "B"
# Each should have its own rate limit
start_time = time.time()
function_a() # Should wait
elapsed = time.time() - start_time
assert elapsed >= 0.4
# function_b should still be able to call without waiting
start_time = time.time()
function_b() # Should not wait much since it has separate limit
elapsed = time.time() - start_time
assert elapsed >= 0.4 # It will also wait since it was called recently
-149
View File
@@ -1,149 +0,0 @@
"""Test the Router symbol detection functionality."""
from unittest.mock import Mock, patch
from common.symbol_router import Router
from common.Symbol import Stock, Coin
class TestRouter:
"""Test Router symbol detection and parsing."""
def setup_method(self):
"""Set up test fixtures."""
# Mock the MarketData and cg_Crypto dependencies
with patch("common.symbol_router.MarketData") as mock_md, patch("common.symbol_router.cg_Crypto") as mock_crypto:
# Mock MarketData
mock_md_instance = Mock()
mock_md.return_value = mock_md_instance
# Mock cg_Crypto
mock_crypto_instance = Mock()
mock_crypto.return_value = mock_crypto_instance
# Create mock symbol list for crypto
import pandas as pd
mock_crypto_instance.symbol_list = pd.DataFrame(
{
"symbol": ["btc", "eth", "ada"],
"id": ["bitcoin", "ethereum", "cardano"],
"name": ["Bitcoin", "Ethereum", "Cardano"],
}
)
self.router = Router()
self.mock_stock = mock_md_instance
self.mock_crypto = mock_crypto_instance
def test_stock_symbol_regex(self):
"""Test that stock symbol regex correctly identifies patterns."""
import re
# Test the regex pattern directly
pattern = self.router.STOCK_REGEX
# Should match
assert re.search(pattern, "$AAPL")
assert re.search(pattern, "What about $TSLA today?")
assert re.search(pattern, "Look at $NVDA and $AMD")
# Should not match crypto patterns
assert not re.search(pattern, "$$BTC")
# Should not match without $
assert not re.search(pattern, "AAPL")
def test_crypto_symbol_regex(self):
"""Test that crypto symbol regex correctly identifies patterns."""
import re
pattern = self.router.CRYPTO_REGEX
# Should match
assert re.search(pattern, "$$BTC")
assert re.search(pattern, "How is $$ETH doing?")
assert re.search(pattern, "$$BTC and $$ETH")
# Should not match single $ patterns
assert not re.search(pattern, "$BTC")
assert not re.search(pattern, "$AAPL")
def test_find_symbols_stock(self):
"""Test stock symbol detection."""
# Mock successful stock lookup
self.mock_stock.symbol_id.return_value = {"ticker": "AAPL", "title": "Apple Inc.", "mkt_cap_rank": 1}
result = self.router.find_symbols("$AAPL")
assert len(result) == 1
assert isinstance(result[0], Stock)
self.mock_stock.symbol_id.assert_called_with("AAPL")
def test_find_symbols_crypto(self):
"""Test crypto symbol detection."""
# Mock crypto dataframe lookup
import pandas as pd
mock_crypto_df = pd.DataFrame({"symbol": ["btc"], "id": ["bitcoin"], "name": ["Bitcoin"], "type_id": ["$$btc"]})
# Set up the mock to return the DataFrame when filtered
self.mock_crypto.symbol_list = pd.DataFrame(
{
"symbol": ["btc", "eth"],
"id": ["bitcoin", "ethereum"],
"name": ["Bitcoin", "Ethereum"],
"type_id": ["$$btc", "$$eth"],
}
)
# Mock the pandas filtering
with patch.object(self.mock_crypto.symbol_list, "str") as mock_str:
mock_str.fullmatch.return_value = mock_crypto_df
mock_crypto_df.empty = False
result = self.router.find_symbols("$$BTC")
assert len(result) == 1
assert isinstance(result[0], Coin)
def test_find_symbols_no_matches(self):
"""Test behavior when no symbols are found."""
result = self.router.find_symbols("Hello world")
assert result == []
def test_find_symbols_invalid_stock(self):
"""Test behavior when stock symbol is not found."""
# Mock failed stock lookup
self.mock_stock.symbol_id.return_value = None
result = self.router.find_symbols("$INVALID")
assert result == []
self.mock_stock.symbol_id.assert_called_with("INVALID")
def test_find_symbols_multiple(self):
"""Test finding multiple symbols."""
# Mock successful lookups for both
self.mock_stock.symbol_id.side_effect = [
{"ticker": "AAPL", "title": "Apple Inc.", "mkt_cap_rank": 1},
{"ticker": "TSLA", "title": "Tesla Inc.", "mkt_cap_rank": 5},
]
result = self.router.find_symbols("$AAPL and $TSLA")
assert len(result) == 2
assert all(isinstance(symbol, Stock) for symbol in result)
def test_trending_weight_tracking(self):
"""Test that trending counts are updated."""
# Mock successful stock lookup
self.mock_stock.symbol_id.return_value = {"ticker": "AAPL", "title": "Apple Inc.", "mkt_cap_rank": 1}
# Clear trending count
self.router.trending_count = {}
result = self.router.find_symbols("$AAPL", trending_weight=2)
assert len(result) == 1
assert "$AAPL" in self.router.trending_count
assert self.router.trending_count["$AAPL"] == 2
-128
View File
@@ -1,128 +0,0 @@
"""Test Symbol, Stock, and Coin class construction."""
from unittest.mock import patch
import pandas as pd
from common.Symbol import Symbol, Stock, Coin
class TestSymbol:
"""Test base Symbol class."""
def test_symbol_creation(self):
"""Test basic Symbol creation."""
symbol = Symbol("TEST")
assert symbol.symbol == "TEST"
assert symbol.id == "TEST"
assert symbol.name == "TEST"
def test_symbol_currency_default(self):
"""Test that currency defaults to USD."""
symbol = Symbol("TEST")
assert symbol.currency == "usd"
class TestStock:
"""Test Stock class functionality."""
def test_stock_creation(self):
"""Test Stock object creation with proper dictionary structure."""
stock_info = {"ticker": "AAPL", "title": "Apple Inc.", "mkt_cap_rank": 1}
stock = Stock(stock_info)
assert stock.symbol == "AAPL"
assert stock.id == "AAPL"
assert stock.name == "Apple Inc."
assert stock.tag == "$AAPL"
assert stock.market_cap_rank == 1
# Verify inheritance
assert isinstance(stock, Symbol)
assert isinstance(stock, Stock)
def test_stock_with_different_data(self):
"""Test Stock creation with different stock data."""
stock_info = {"ticker": "TSLA", "title": "Tesla, Inc.", "mkt_cap_rank": 5}
stock = Stock(stock_info)
assert stock.symbol == "TSLA"
assert stock.id == "TSLA"
assert stock.name == "Tesla, Inc."
assert stock.tag == "$TSLA"
assert stock.market_cap_rank == 5
def test_stock_tag_format(self):
"""Test that stock tags are properly formatted with $ prefix."""
stock_info = {"ticker": "NVDA", "title": "NVIDIA Corporation", "mkt_cap_rank": 3}
stock = Stock(stock_info)
assert stock.tag == "$NVDA"
assert stock.tag.startswith("$")
class TestCoin:
"""Test Coin class functionality."""
def test_coin_creation_single_row(self):
"""Test Coin object creation with single row DataFrame."""
coin_data = pd.DataFrame({"symbol": ["btc"], "id": ["bitcoin"], "name": ["Bitcoin"], "type_id": ["$$btc"]})
coin = Coin(coin_data)
assert coin.symbol == "btc"
assert coin.id == "bitcoin"
assert coin.name == "Bitcoin"
assert coin.tag == "$$BTC"
# Verify inheritance
assert isinstance(coin, Symbol)
assert isinstance(coin, Coin)
def test_coin_creation_ethereum(self):
"""Test Coin creation with Ethereum data."""
coin_data = pd.DataFrame({"symbol": ["eth"], "id": ["ethereum"], "name": ["Ethereum"], "type_id": ["$$eth"]})
coin = Coin(coin_data)
assert coin.symbol == "eth"
assert coin.id == "ethereum"
assert coin.name == "Ethereum"
assert coin.tag == "$$ETH"
def test_coin_multiple_rows_takes_first(self):
"""Test that when multiple rows exist, only the first is used."""
coin_data = pd.DataFrame(
{
"symbol": ["btc", "btc2"],
"id": ["bitcoin", "bitcoin-cash"],
"name": ["Bitcoin", "Bitcoin Cash"],
"type_id": ["$$btc", "$$bch"],
}
)
# Mock the head method to return first row
with patch.object(coin_data, "head") as mock_head:
first_row = pd.DataFrame({"symbol": ["btc"], "id": ["bitcoin"], "name": ["Bitcoin"], "type_id": ["$$btc"]})
mock_head.return_value = first_row
coin = Coin(coin_data)
mock_head.assert_called_once_with(1)
# The coin should use the first row data
assert coin.symbol == "btc"
assert coin.id == "bitcoin"
def test_coin_tag_uppercase(self):
"""Test that coin tags are properly uppercased."""
coin_data = pd.DataFrame({"symbol": ["ada"], "id": ["cardano"], "name": ["Cardano"], "type_id": ["$$ada"]})
coin = Coin(coin_data)
assert coin.tag == "$$ADA"
assert coin.tag.isupper()
def test_coin_dataframe_access(self):
"""Test that coin properly accesses DataFrame values."""
coin_data = pd.DataFrame({"symbol": ["dot"], "id": ["polkadot"], "name": ["Polkadot"], "type_id": ["$$dot"]})
coin = Coin(coin_data)
# Test that values are extracted correctly from DataFrame
assert isinstance(coin.symbol, str)
assert isinstance(coin.id, str)
assert isinstance(coin.name, str)
assert isinstance(coin.tag, str)