joungmin/openclaw-workspace
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add: Real market data with yfinance

Browse Source 
- stock_tracker.py now uses yfinance for real prices
- get_market_indices(): KOSPI, S&P 500, NASDAQ, DOW
- get_crypto_price(): BTC, ETH, SOL with 52W range
- CLI commands: 'python stock_tracker.py market' and 'crypto'

Features:
- Live prices from Yahoo Finance
- Market indices tracking
- Cryptocurrency prices
- 52-week high/low
- Daily change percentage

Example usage:
  python stock_tracker.py market   # Show indices
  python stock_tracker.py crypto --symbol BTC  # BTC price
This commit is contained in:
Joungmin
2026-02-19 04:11:30 +09:00
parent 495113a83d
commit 63e7a2ba32
1 changed files with 139 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

155
stock_tracker.py
View File

@@ -17,6 +17,14 @@ from typing import List, Dict, Optional, Tuple
from enum import Enum
import requests
# Try import yfinance for real market data
try:
import yfinance as yf
YFINANCE_AVAILABLE = True
except ImportError:
YFINANCE_AVAILABLE = False
print("⚠️ yfinance not installed. Run: pip install yfinance")
# Configuration
DATA_DIR = os.environ.get('PORTFOLIO_DATA_DIR', '/tmp/portfolio')
os.makedirs(DATA_DIR, exist_ok=True)
@@ -130,22 +138,46 @@ class StockTracker:
# ============== Price Fetching ==============
def fetch_price(self, symbol: str) -> Optional[PriceData]:
"""Fetch current price for a symbol"""
# Placeholder - would use real API
# For demo, generate mock data
import random
mock_price = random.uniform(10000, 500000)
mock_change = random.uniform(-5, 5)
return PriceData(
symbol=symbol,
current_price=mock_price,
change_percent=mock_change,
high_52w=mock_price * 1.2,
low_52w=mock_price * 0.8,
volume=random.uniform(100000, 10000000),
updated_at=datetime.datetime.now().isoformat()
)
"""Fetch current price for a symbol using yfinance"""
if YFINANCE_AVAILABLE:
try:
# Add .KS for Korean stocks, normal for others
ticker = yf.Ticker(symbol)
info = ticker.info
current_price = info.get('currentPrice', info.get('regularMarketPrice', 0))
change_percent = info.get('regularMarketChangePercent', 0) * 100
high_52w = info.get('fiftyTwoWeekHigh', 0)
low_52w = info.get('fiftyTwoWeekLow', 0)
volume = info.get('volume', 0)
return PriceData(
symbol=symbol,
current_price=current_price,
change_percent=change_percent,
high_52w=high_52w,
low_52w=low_52w,
volume=volume,
updated_at=datetime.datetime.now().isoformat()
)
except Exception as e:
print(f"Error fetching {symbol}: {e}")
return None
else:
# Fallback to mock data if yfinance not available
import random
mock_price = random.uniform(10000, 500000)
mock_change = random.uniform(-5, 5)
return PriceData(
symbol=symbol,
current_price=mock_price,
change_percent=mock_change,
high_52w=mock_price * 1.2,
low_52w=mock_price * 0.8,
volume=random.uniform(100000, 10000000),
updated_at=datetime.datetime.now().isoformat()
)
def update_prices(self) -> Dict[str, PriceData]:
"""Update all prices"""
@@ -259,6 +291,74 @@ class StockTracker:
return ((price.current_price - pos.avg_cost) / pos.avg_cost) * 100
return 0
# ============== Crypto & Market Data ==============
def get_crypto_price(self, symbol: str = "BTC") -> Optional[PriceData]:
"""Fetch cryptocurrency price using yfinance"""
if not YFINANCE_AVAILABLE:
return None
try:
ticker = yf.Ticker(f"{symbol}-USD")
hist = ticker.history(period='1d')
hist_1d = ticker.history(period='2d') # Get 2 days for change
if hist.empty:
return None
current_price = hist['Close'].iloc[-1]
prev_close = hist_1d['Close'].iloc[0] if len(hist_1d) > 1 else current_price
change_percent = ((current_price - prev_close) / prev_close) * 100 if prev_close > 0 else 0
# Get 52-week data
hist_52w = ticker.history(period='1y')
high_52w = hist_52w['High'].max() if not hist_52w.empty else current_price * 1.2
low_52w = hist_52w['Low'].min() if not hist_52w.empty else current_price * 0.8
return PriceData(
symbol=symbol,
current_price=current_price,
change_percent=change_percent,
high_52w=high_52w,
low_52w=low_52w,
volume=hist['Volume'].iloc[-1] if 'Volume' in hist.columns else 0,
updated_at=datetime.datetime.now().isoformat()
)
except Exception as e:
print(f"Error fetching crypto {symbol}: {e}")
return None
def get_market_indices(self) -> Dict[str, Dict]:
"""Fetch major market indices using yfinance"""
indices = {
'KOSPI': '^KS11',
'KOSDAQ': '^KOSDAQ',
'S&P 500': '^GSPC',
'NASDAQ': '^IXIC',
'DOW': '^DJI'
}
result = {}
if YFINANCE_AVAILABLE:
for name, ticker in indices.items():
try:
t = yf.Ticker(ticker)
hist = t.history(period='1d')
hist_1d = t.history(period='2d') # Get 2 days for change calculation
if not hist.empty:
current = hist['Close'].iloc[-1]
prev_close = hist_1d['Close'].iloc[0] if len(hist_1d) > 1 else current
change = ((current - prev_close) / prev_close) * 100 if prev_close > 0 else 0
result[name] = {'price': current, 'change': change}
else:
result[name] = {'price': 0, 'change': 0}
except Exception as e:
print(f"Error fetching {name}: {e}")
result[name] = {'price': 0, 'change': 0}
return result
# ============== Reporting ==============
def generate_daily_report(self) -> str:
@@ -346,6 +446,13 @@ def main():
# Weekly report
subparsers.add_parser('weekly', help='Generate weekly report')
# Crypto price
crypto_parser = subparsers.add_parser('crypto', help='Get crypto price')
crypto_parser.add_argument('--symbol', default='BTC', help='Crypto symbol (BTC, ETH, etc.)')
# Market indices
subparsers.add_parser('market', help='Show market indices')
args = parser.parse_args()
tracker = StockTracker()
@@ -374,6 +481,22 @@ def main():
elif args.command == 'weekly':
print(tracker.generate_weekly_report())
elif args.command == 'crypto':
price = tracker.get_crypto_price(args.symbol)
if price:
emoji = "🟢" if price.change_percent > 0 else "🔴"
print(f"\n{emoji} {args.symbol}: ${price.current_price:,.2f} ({price.change_percent:+.2f}%)")
print(f" 52W Range: ${price.low_52w:,.2f} - ${price.high_52w:,.2f}")
else:
print("❌ yfinance not available. Install: pip install yfinance")
elif args.command == 'market':
indices = tracker.get_market_indices()
print("\n📈 Market Indices")
for name, data in indices.items():
emoji = "🟢" if data['change'] > 0 else "🔴"
print(f" {emoji} {name}: {data['price']:,.2f} ({data['change']:+.2f}%)")
else:
parser.print_help()