yirugi
/
yfinance_plus
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

first commit

Browse Source
main
I Luk Kim 1 year ago
parent 790e6ead0d
commit 85855b6ec6
7 changed files with 1289 additions and 0 deletions
Show Stats Download Patch File Download Diff File

284
.gitignore vendored
Unescape Escape View File

@ -0,0 +1,284 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# poetry
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
#poetry.lock
# pdm
# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
#pdm.lock
# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
# in version control.
# https://pdm.fming.dev/#use-with-ide
.pdm.toml
# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
# PyCharm
# JetBrains specific template is maintained in a separate JetBrains.gitignore that can
# be added to the global gitignore or merged into this project gitignore. For a PyCharm
# project, it is recommended to use the following settings:
#
# Editor | File Types | Ignored Files and Folders
#
# Import from .gitignore
.idea/
# VSCode
.vscode/
# macOS
.DS_Store
.AppleDouble
.LSOverride
# Icon must end with two \r
Icon
# Thumbnails
._*
# Files that might appear in the root of a volume
.DocumentRevisions-V100
.fseventsd
.Spotlight-V100
.TemporaryItems
.Trashes
.VolumeIcon.icns
.com.apple.timemachine.donotpresent
# Directories potentially created on remote AFP share
.AppleDB
.AppleDesktop
Network Trash Folder
Temporary Items
.apdisk
# Windows
Thumbs.db
Thumbs.db:encryptable
ehthumbs.db
ehthumbs_vista.db
*.tmp
*.temp
*.bak
*.swp
*.swo
*~.nib
local.properties
.settings/
.loadpath
.recommenders
target/
.metadata
bin/
tmp/
*.tmp
*.bak
*.swp
*~.nib
local.properties
.settings/
.loadpath
.recommenders
# Project specific files
test_*.py
example*.py
check_*.py
comprehensive_*.py
*_old.py
debug*.py
temp*.py
# Cache directories
.yfinance_cache/
cache/
*.cache
# Log files
*.log
logs/
# Temporary files
temp/
tmp/
*.tmp
*.temp
# Data files
*.csv
*.json
*.pickle
*.pkl
data/
# Jupyter notebooks (optional - uncomment if you don't want to track notebooks)
# *.ipynb
# Environment variables
.env.local
.env.development
.env.test
.env.production
# Coverage reports
htmlcov/
.coverage
coverage.xml
# Documentation build
docs/_build/
docs/build/
# Virtual environments
venv*/
env*/
# OS generated files
Desktop.ini
$RECYCLE.BIN/
*.cab
*.msi
*.msix
*.msm
*.msp
*.lnk

21
LICENSE
Unescape Escape View File

@ -0,0 +1,21 @@
MIT License
Copyright (c) 2024 YFinance Enhanced Team
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

7
MANIFEST.in
Unescape Escape View File

@ -0,0 +1,7 @@
include README.md
include requirements.txt
include LICENSE
recursive-include . *.py
recursive-exclude . __pycache__
recursive-exclude . *.pyc
recursive-exclude . .DS_Store

13
requirements.txt
Unescape Escape View File

@ -0,0 +1,13 @@
yfinance>=0.2.65
curl_cffi>=0.5.0
pandas>=1.3.0
numpy>=1.20.0
beautifulsoup4>=4.9.0
requests>=2.25.0
multitasking>=0.0.10
frozendict>=2.0.0
peewee>=3.14.0
platformdirs>=2.0.0
protobuf>=3.19.0
pytz>=2021.1
websockets>=10.0

67
setup.py
Unescape Escape View File

@ -0,0 +1,67 @@
#!/usr/bin/env python3
from setuptools import setup, find_packages
import os
# Read the README file
def read_readme():
with open("README.md", "r", encoding="utf-8") as fh:
return fh.read()
# Read requirements
def read_requirements():
with open("requirements.txt", "r", encoding="utf-8") as fh:
return [line.strip() for line in fh if line.strip() and not line.startswith("#")]
setup(
name="yfinance-plus",
version="1.0.0",
author="YFinance Plus Team",
author_email="example@example.com",
description="Enhanced yfinance wrapper with improved rate limit handling and intelligent caching",
long_description=read_readme(),
long_description_content_type="text/markdown",
url="https://gitea.yirugi.synology.me/yirugi/yfinance_plus.git",
py_modules=["yfinance_enhanced", "yfinance_enhanced_cli"],
classifiers=[
"Development Status :: 4 - Beta",
"Intended Audience :: Developers",
"Intended Audience :: Financial and Insurance Industry",
"License :: OSI Approved :: MIT License",
"Operating System :: OS Independent",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Topic :: Office/Business :: Financial",
"Topic :: Office/Business :: Financial :: Investment",
"Topic :: Scientific/Engineering :: Information Analysis",
"Topic :: Software Development :: Libraries :: Python Modules",
],
python_requires=">=3.8",
install_requires=read_requirements(),
extras_require={
"dev": [
"pytest>=6.0",
"pytest-cov",
"black",
"flake8",
"mypy",
],
},
entry_points={
"console_scripts": [
"yfp=yfinance_enhanced_cli:main",
],
},
keywords="finance, yfinance, yahoo finance, stocks, market data, plus, rate limit, caching",
project_urls={
"Bug Reports": "https://gitea.yirugi.synology.me/yirugi/yfinance_plus/issues",
"Source": "https://gitea.yirugi.synology.me/yirugi/yfinance_plus.git",
"Documentation": "https://gitea.yirugi.synology.me/yirugi/yfinance_plus/src/branch/main/README.md",
},
include_package_data=True,
zip_safe=False,
)

592
yfinance_enhanced.py
Unescape Escape View File

@ -0,0 +1,592 @@
#!/usr/bin/env python3
"""
Enhanced yfinance wrapper with improved rate limit handling and intelligent caching.
Uses dynamic proxy pattern to wrap all yfinance functionality automatically.
"""
import os
import time
import random
import threading
import pickle
import hashlib
from pathlib import Path
from typing import Dict, Optional, Any, Union, List
from dataclasses import dataclass
from functools import wraps
import logging
from datetime import datetime, timedelta
import yfinance as yf
from curl_cffi import requests
import pandas as pd
@dataclass
class RequestConfig:
"""Configuration for enhanced requests"""
max_retries: int = 3
base_delay: float = 1.0
max_delay: float = 60.0
jitter: bool = True
user_agents: list = None
enable_cache: bool = True
cache_dir: str = None
def __post_init__(self):
if self.user_agents is None:
self.user_agents = [
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.2.1 Safari/605.1.15',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36 Edg/119.0.0.0'
]
if self.cache_dir is None:
self.cache_dir = os.path.expanduser('~/.yfinance_cache')
class HistoricalDataCache:
"""Intelligent cache for historical data"""
def __init__(self, cache_dir: str):
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(parents=True, exist_ok=True)
self._lock = threading.Lock()
def _get_cache_key(self, symbol: str, period: str, interval: str, start: str = None, end: str = None) -> str:
"""Generate cache key for historical data request"""
key_data = f"{symbol}_{period}_{interval}_{start}_{end}"
return hashlib.md5(key_data.encode()).hexdigest()
def _get_cache_file(self, cache_key: str) -> Path:
"""Get cache file path"""
return self.cache_dir / f"{cache_key}.pkl"
def _normalize_period_to_days(self, period: str) -> int:
"""Convert period string to approximate days"""
period_map = {
'1d': 1, '2d': 2, '5d': 5, '1mo': 30, '3mo': 90, '6mo': 180,
'1y': 365, '2y': 730, '5y': 1825, '10y': 3650, 'ytd': 365, 'max': 36500
}
return period_map.get(period.lower(), 30)
def get(self, symbol: str, period: str, interval: str, start: str = None, end: str = None) -> Optional[pd.DataFrame]:
"""Get cached historical data if available and valid"""
with self._lock:
try:
# First check exact match
cache_key = self._get_cache_key(symbol, period, interval, start, end)
cache_file = self._get_cache_file(cache_key)
if cache_file.exists():
with open(cache_file, 'rb') as f:
cache_data = pickle.load(f)
# Check if cache is still valid (not older than 1 hour for recent data)
cache_time = datetime.fromtimestamp(cache_file.stat().st_mtime)
if datetime.now() - cache_time < timedelta(hours=1):
return cache_data['data']
# Look for longer period cache that might contain requested data
requested_days = self._normalize_period_to_days(period)
for cache_file in self.cache_dir.glob(f"*.pkl"):
try:
with open(cache_file, 'rb') as f:
cache_data = pickle.load(f)
# Check if this cache contains data for the same symbol and interval
if (cache_data['symbol'] == symbol and
cache_data['interval'] == interval and
cache_data['period_days'] >= requested_days):
# Check cache validity
cache_time = datetime.fromtimestamp(cache_file.stat().st_mtime)
if datetime.now() - cache_time < timedelta(hours=1):
# Extract requested period from cached data
data = cache_data['data']
if len(data) >= requested_days:
return data.tail(min(len(data), requested_days * 2)) # Buffer for weekends
except:
continue
return None
except Exception as e:
logging.warning(f"Cache read error: {e}")
return None
def store(self, symbol: str, period: str, interval: str, data: pd.DataFrame, start: str = None, end: str = None):
"""Store historical data in cache"""
with self._lock:
try:
cache_key = self._get_cache_key(symbol, period, interval, start, end)
cache_file = self._get_cache_file(cache_key)
cache_data = {
'symbol': symbol,
'period': period,
'period_days': self._normalize_period_to_days(period),
'interval': interval,
'start': start,
'end': end,
'data': data,
'cached_at': datetime.now()
}
with open(cache_file, 'wb') as f:
pickle.dump(cache_data, f)
except Exception as e:
logging.warning(f"Cache store error: {e}")
def clear(self):
"""Clear all cached data"""
with self._lock:
for cache_file in self.cache_dir.glob("*.pkl"):
try:
cache_file.unlink()
except:
pass
def get_cache_info(self) -> Dict[str, Any]:
"""Get cache statistics"""
with self._lock:
cache_files = list(self.cache_dir.glob("*.pkl"))
total_size = sum(f.stat().st_size for f in cache_files)
return {
'cache_dir': str(self.cache_dir),
'file_count': len(cache_files),
'total_size_mb': total_size / (1024 * 1024),
'files': [{
'name': f.name,
'size_kb': f.stat().st_size / 1024,
'modified': datetime.fromtimestamp(f.stat().st_mtime)
} for f in cache_files]
}
class EnhancedYFinance:
"""Enhanced yfinance wrapper with better rate limit handling and caching"""
def __init__(self, config: RequestConfig = None):
self.config = config or RequestConfig()
self._session = None
self._lock = threading.Lock()
self._last_request_time = 0
self._request_count = 0
self._session_created_time = time.time()
# Setup caching
if self.config.enable_cache:
self._cache = HistoricalDataCache(self.config.cache_dir)
else:
self._cache = None
# Setup logging
self.logger = logging.getLogger(__name__)
if not self.logger.handlers:
handler = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
self.logger.addHandler(handler)
self.logger.setLevel(logging.WARNING) # Reduce log verbosity
# Suppress yfinance HTTP error logs
yf_logger = logging.getLogger('yfinance')
yf_logger.setLevel(logging.CRITICAL)
def _create_enhanced_session(self) -> requests.Session:
"""Create a session with browser-like headers and behavior"""
session = requests.Session(impersonate="chrome")
# Enhanced headers to mimic real browser behavior
browser_headers = {
'User-Agent': random.choice(self.config.user_agents),
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
'Accept-Language': 'en-US,en;q=0.9,ko;q=0.8,ja;q=0.7',
'Accept-Encoding': 'gzip, deflate, br',
'Cache-Control': 'no-cache',
'Pragma': 'no-cache',
'Sec-Ch-Ua': '"Not_A Brand";v="8", "Chromium";v="120", "Google Chrome";v="120"',
'Sec-Ch-Ua-Mobile': '?0',
'Sec-Ch-Ua-Platform': '"macOS"',
'Sec-Fetch-Dest': 'document',
'Sec-Fetch-Mode': 'navigate',
'Sec-Fetch-Site': 'cross-site',
'Sec-Fetch-User': '?1',
'Upgrade-Insecure-Requests': '1',
'Connection': 'keep-alive',
'DNT': '1',
'Origin': 'https://finance.yahoo.com',
'Referer': 'https://finance.yahoo.com/',
}
session.headers.update(browser_headers)
# Add realistic Yahoo Finance cookies
current_time = int(time.time())
session.cookies.update({
'A1': f'd=AQABBC{random.randint(100000, 999999)}YLoCIgEBwQJ7vgAB&S=AQAAAg',
'A1S': f'd=AQABBC{random.randint(100000, 999999)}YLoCIgEBwQJ7vgAB&S=AQAAAg',
'A3': f'd=AQABBC{random.randint(100000, 999999)}YLoCIgEBwQJ7vgAB&S=AQAAAg',
'GUC': f'AQEBAQFm{random.randint(1000, 9999)}k0L',
'B': f'c={random.randint(1000000, 9999999)}&b=3&s=4u',
'cmp': f't={current_time}&j=0&u=1---',
'EuConsent': 'CP-r9cAP-r9cAAOACKENAoEgAAAAAAAAACiQAAAAAAAA',
})
return session
@property
def session(self) -> requests.Session:
"""Get or create enhanced session"""
with self._lock:
if self._session is None or self._should_refresh_session():
self._session = self._create_enhanced_session()
self._session_created_time = time.time()
self._request_count = 0
self.logger.debug("Created new enhanced session")
return self._session
def _should_refresh_session(self) -> bool:
"""Check if session should be refreshed"""
session_age = time.time() - self._session_created_time
return (session_age > 600 or # 10 minutes
self._request_count > 50) # 50 requests
def _rate_limit_delay(self):
"""Apply intelligent rate limiting"""
with self._lock:
current_time = time.time()
time_since_last = current_time - self._last_request_time
# Minimum delay based on request frequency
min_delay = 0.1 if self._request_count < 10 else 0.2
if time_since_last < min_delay:
delay = min_delay - time_since_last
if self.config.jitter:
delay += random.uniform(0, delay * 0.5)
time.sleep(delay)
self._last_request_time = time.time()
self._request_count += 1
def _retry_with_backoff(self, func, *args, **kwargs):
"""Execute function with exponential backoff retry"""
last_exception = None
for attempt in range(self.config.max_retries + 1):
try:
self._rate_limit_delay()
return func(*args, **kwargs)
except Exception as e:
last_exception = e
error_str = str(e).lower()
# Check if it's a rate limit or 401 error
if ("rate limit" in error_str or "429" in error_str or
"401" in error_str or "unauthorized" in error_str):
if attempt < self.config.max_retries:
delay = min(
self.config.base_delay * (2 ** attempt),
self.config.max_delay
)
if self.config.jitter:
delay += random.uniform(0, delay * 0.3)
if "401" in error_str:
self.logger.debug(f"401 error, refreshing session and retrying in {delay:.2f}s (attempt {attempt + 1})")
else:
self.logger.warning(f"Rate limit hit, retrying in {delay:.2f}s (attempt {attempt + 1})")
time.sleep(delay)
# Refresh session on auth/rate limit error
with self._lock:
self._session = None
continue
else:
# Non-rate-limit error, re-raise immediately
raise e
# All retries exhausted
raise last_exception
def get_ticker(self, symbol: str) -> 'EnhancedTicker':
"""Get enhanced ticker with improved rate limiting"""
return EnhancedTicker(symbol, self)
class EnhancedTicker:
"""
Enhanced ticker wrapper using dynamic proxy pattern.
Automatically wraps ALL yfinance ticker functionality.
"""
# Methods that should be cached
_CACHEABLE_METHODS = {'history'}
# Methods that should NOT be cached (real-time or frequently changing data)
_NON_CACHEABLE_METHODS = {'news', 'get_news', 'fast_info'}
def __init__(self, symbol: str, enhanced_yf: EnhancedYFinance):
self.symbol = symbol.upper()
self.enhanced_yf = enhanced_yf
self._yf_ticker = None
@property
def yf_ticker(self):
"""Get yfinance ticker with enhanced session"""
if self._yf_ticker is None:
self._yf_ticker = yf.Ticker(self.symbol, session=self.enhanced_yf.session)
return self._yf_ticker
def __getattr__(self, name):
"""
Dynamic proxy: automatically wrap any yfinance method or property
"""
# Get the original attribute from yfinance ticker
original_attr = getattr(self.yf_ticker, name)
# If it's a method/function, wrap it with enhanced functionality
if callable(original_attr):
if name == 'history':
# Special handling for history method with caching
return self._enhanced_history
elif name in self._NON_CACHEABLE_METHODS:
# Methods that should not be cached
return self._wrap_method_no_cache(original_attr)
else:
# All other methods get standard enhancement
return self._wrap_method(original_attr)
else:
# For properties, wrap them with retry logic
return self._wrap_property(name)
def _enhanced_history(self, period="1mo", interval="1d", start=None, end=None, **kwargs):
"""Enhanced history method with caching"""
# Check cache first for historical data (not real-time)
if (self.enhanced_yf._cache and
not kwargs.get('prepost', False)): # Don't cache extended hours data
cached_data = self.enhanced_yf._cache.get(
self.symbol, period, interval,
str(start) if start else None,
str(end) if end else None
)
if cached_data is not None and not cached_data.empty:
self.enhanced_yf.logger.debug(f"Using cached data for {self.symbol}")
return cached_data
# Fetch fresh data
def _get_history():
return self.yf_ticker.history(
period=period, interval=interval,
start=start, end=end, **kwargs
)
data = self.enhanced_yf._retry_with_backoff(_get_history)
# Cache the data if successful and caching is enabled
if (self.enhanced_yf._cache and
data is not None and not data.empty and
not kwargs.get('prepost', False)):
self.enhanced_yf._cache.store(
self.symbol, period, interval, data,
str(start) if start else None,
str(end) if end else None
)
return data
def _wrap_method(self, method):
"""Wrap a method with enhanced error handling and rate limiting"""
@wraps(method)
def wrapped(*args, **kwargs):
def _call_method():
return method(*args, **kwargs)
return self.enhanced_yf._retry_with_backoff(_call_method)
return wrapped
def _wrap_method_no_cache(self, method):
"""Wrap a method with enhanced error handling but no caching"""
@wraps(method)
def wrapped(*args, **kwargs):
def _call_method():
return method(*args, **kwargs)
return self.enhanced_yf._retry_with_backoff(_call_method)
return wrapped
def _wrap_property(self, name):
"""Wrap a property with enhanced error handling"""
def _get_property():
return getattr(self.yf_ticker, name)
return self.enhanced_yf._retry_with_backoff(_get_property)
# Additional utility methods
def get_cache_info(self) -> Dict[str, Any]:
"""Get cache information for this ticker"""
if self.enhanced_yf._cache:
return self.enhanced_yf._cache.get_cache_info()
return {'cache_enabled': False}
def clear_cache(self):
"""Clear cache for this ticker"""
if self.enhanced_yf._cache:
self.enhanced_yf._cache.clear()
def download(tickers: Union[str, list],
period: str = "1mo",
interval: str = "1d",
**kwargs) -> pd.DataFrame:
"""Enhanced download function with better rate limiting"""
config = RequestConfig(
max_retries=5,
base_delay=1.0,
max_delay=120.0
)
enhanced_yf = EnhancedYFinance(config)
def _download():
return yf.download(
tickers=tickers,
period=period,
interval=interval,
session=enhanced_yf.session,
**kwargs
)
return enhanced_yf._retry_with_backoff(_download)
# Global configuration
_global_config = RequestConfig()
def set_config(**kwargs):
"""Set global configuration for enhanced yfinance"""
global _global_config
for key, value in kwargs.items():
if hasattr(_global_config, key):
setattr(_global_config, key, value)
else:
raise ValueError(f"Unknown configuration option: {key}")
def get_config() -> RequestConfig:
"""Get current global configuration"""
return _global_config
# Convenience functions with full yfinance API compatibility
def Ticker(symbol: str, session=None, proxy=None) -> EnhancedTicker:
"""Create enhanced ticker (fully compatible with yfinance.Ticker)"""
config = RequestConfig(
max_retries=_global_config.max_retries,
base_delay=_global_config.base_delay,
max_delay=_global_config.max_delay,
jitter=_global_config.jitter,
user_agents=_global_config.user_agents,
enable_cache=_global_config.enable_cache,
cache_dir=_global_config.cache_dir
)
enhanced_yf = EnhancedYFinance(config)
# Override session if provided (for compatibility)
if session is not None:
enhanced_yf._session = session
return enhanced_yf.get_ticker(symbol)
class EnhancedTickers:
"""Enhanced Tickers class with full yfinance compatibility"""
def __init__(self, tickers, session=None, proxy=None):
# Parse tickers like original yfinance
if isinstance(tickers, str):
tickers = tickers.replace(',', ' ').split()
elif not isinstance(tickers, list):
tickers = list(tickers)
self.symbols = [ticker.upper() for ticker in tickers]
self.tickers = {symbol: Ticker(symbol, session, proxy) for symbol in self.symbols}
# Store original yfinance Tickers for method delegation
self._yf_tickers = yf.Tickers(self.symbols, session=session)
def __repr__(self):
return f"yfinance_enhanced.Tickers object <{','.join(self.symbols)}>"
def __getattr__(self, name):
"""Delegate any missing methods to original yfinance Tickers"""
return getattr(self._yf_tickers, name)
def download(self, *args, **kwargs):
"""Enhanced download for multiple tickers"""
return download(self.symbols, *args, **kwargs)
# Additional yfinance compatibility functions
def Tickers(tickers, session=None, proxy=None) -> EnhancedTickers:
"""Create enhanced tickers (fully compatible with yfinance.Tickers)"""
return EnhancedTickers(tickers, session, proxy)
# Cache management functions
def clear_cache():
"""Clear all cached data"""
cache = HistoricalDataCache(_global_config.cache_dir)
cache.clear()
def get_cache_info():
"""Get cache information"""
cache = HistoricalDataCache(_global_config.cache_dir)
return cache.get_cache_info()
# Make the module work exactly like yfinance
from yfinance import __version__
# Export main classes and functions for full compatibility
__all__ = [
'Ticker', 'Tickers', 'download', 'set_config', 'get_config',
'clear_cache', 'get_cache_info', 'RequestConfig',
'EnhancedTicker', 'EnhancedTickers', 'EnhancedYFinance', 'HistoricalDataCache'
]
# Example usage
if __name__ == "__main__":
# Setup logging
logging.basicConfig(level=logging.INFO)
# Create enhanced ticker
ticker = Ticker("AAPL")
try:
# Test all yfinance functionality automatically
print("Testing enhanced yfinance with automatic wrapping...")
# All these should work automatically via __getattr__
print(f"Info: {ticker.info.get('longName', 'N/A')}")
print(f"History: {len(ticker.history(period='5d'))} rows")
print(f"Dividends: {len(ticker.dividends)} records")
print(f"Splits: {len(ticker.splits)} records")
print(f"Major holders: {ticker.major_holders.shape if hasattr(ticker.major_holders, 'shape') else 'Available'}")
print(f"Recommendations: {ticker.recommendations.shape if hasattr(ticker.recommendations, 'shape') else 'Available'}")
print(f"News: {len(ticker.news) if ticker.news else 0} articles")
print("✅ All yfinance functionality works automatically!")
except Exception as e:
print(f"Error: {e}")
import traceback
traceback.print_exc()

305
yfinance_enhanced_cli.py
Unescape Escape View File

@ -0,0 +1,305 @@
#!/usr/bin/env python3
"""
Command Line Interface for yfinance-enhanced
Provides cache management and utility functions
"""
import argparse
import sys
import json
import pickle
from pathlib import Path
from datetime import datetime
try:
from yfinance_enhanced import get_cache_info, clear_cache, HistoricalDataCache, RequestConfig
except ImportError:
print("Error: yfinance_enhanced module not found. Please install it first.")
sys.exit(1)
def format_size(size_bytes):
"""Format size in bytes to human readable format"""
if size_bytes == 0:
return "0 B"
for unit in ['B', 'KB', 'MB', 'GB']:
if size_bytes < 1024.0:
return f"{size_bytes:.1f} {unit}"
size_bytes /= 1024.0
return f"{size_bytes:.1f} TB"
def get_cache_file_details(cache_dir, filename):
"""Load and analyze cache file to get details about stored data"""
try:
cache_file = Path(cache_dir) / filename
if not cache_file.exists():
return None
with open(cache_file, 'rb') as f:
cache_data = pickle.load(f)
# Extract details from cache data
details = {
'symbol': cache_data.get('symbol', 'Unknown'),
'period': cache_data.get('period', 'Unknown'),
'interval': cache_data.get('interval', '1d'),
'rows': len(cache_data.get('data', [])),
'date_range': None
}
# Try to get date range from the data
data = cache_data.get('data')
if data is not None and hasattr(data, 'index') and len(data) > 0:
try:
start_date = data.index[0].strftime('%Y-%m-%d')
end_date = data.index[-1].strftime('%Y-%m-%d')
details['date_range'] = f"{start_date} to {end_date}"
except:
pass
return details
except Exception as e:
# If we can't read the cache file, return minimal info
return {
'symbol': 'Error',
'period': f'({str(e)[:20]}...)',
'interval': '?',
'rows': 0,
'date_range': None
}
def cmd_cache_info(args):
"""Show cache information"""
try:
info = get_cache_info()
print("🗂️ YFinance Plus Cache Information")
print("=" * 50)
print(f"Cache Directory: {info['cache_dir']}")
print(f"Number of Files: {info['file_count']}")
print(f"Total Size: {format_size(info['total_size_mb'] * 1024 * 1024)}")
if info['file_count'] > 0:
print(f"\n📁 Cache Files:")
print("-" * 100)
sorted_files = sorted(info['files'], key=lambda x: x['modified'], reverse=True)
for file_info in sorted_files[:args.limit if args.limit else len(sorted_files)]:
size_str = format_size(file_info['size_kb'] * 1024)
modified_str = file_info['modified'].strftime('%Y-%m-%d %H:%M:%S')
# Try to load and analyze cache file content
cache_details = get_cache_file_details(info['cache_dir'], file_info['name'])
print(f" 📄 {file_info['name'][:32]:<32} {size_str:>8} {modified_str}")
if cache_details:
print(f" Symbol: {cache_details['symbol']:<8} Period: {cache_details['period']:<6} "
f"Interval: {cache_details['interval']:<4} Rows: {cache_details['rows']}")
if cache_details['date_range']:
print(f" Date Range: {cache_details['date_range']}")
else:
print(f" ⚠️ Could not read cache file details")
print()
if args.limit and len(sorted_files) > args.limit:
print(f" ... and {len(sorted_files) - args.limit} more files")
else:
print("\n💡 No cache files found")
except Exception as e:
print(f"❌ Error getting cache info: {e}")
sys.exit(1)
def cmd_cache_clear(args):
"""Clear cache"""
if not args.force:
try:
info = get_cache_info()
if info['file_count'] > 0:
response = input(f"⚠️ This will delete {info['file_count']} cache files "
f"({format_size(info['total_size_mb'] * 1024 * 1024)}). "
f"Continue? [y/N]: ")
if response.lower() not in ['y', 'yes']:
print("❌ Operation cancelled")
return
except Exception:
pass
try:
clear_cache()
print("✅ Cache cleared successfully")
except Exception as e:
print(f"❌ Error clearing cache: {e}")
sys.exit(1)
def cmd_config_show(args):
"""Show current configuration"""
config = RequestConfig()
print("⚙️ YFinance Plus Configuration")
print("=" * 40)
print(f"Max Retries: {config.max_retries}")
print(f"Base Delay: {config.base_delay}s")
print(f"Max Delay: {config.max_delay}s")
print(f"Jitter Enabled: {config.jitter}")
print(f"Cache Enabled: {config.enable_cache}")
print(f"Cache Directory: {config.cache_dir}")
print(f"User Agents: {len(config.user_agents)} configured")
def cmd_test_connection(args):
"""Test connection to Yahoo Finance"""
print("🔗 Testing connection to Yahoo Finance...")
try:
# Import here to avoid circular imports
from yfinance_enhanced import Ticker
print(" Creating test ticker...")
ticker = Ticker("AAPL")
print(" Fetching basic info...")
info = ticker.info
company_name = info.get('longName', 'Unknown')
print(" Fetching historical data...")
hist = ticker.history(period="5d")
print(f"✅ Connection test successful!")
print(f" Company: {company_name}")
print(f" Historical data points: {len(hist)}")
if ticker.enhanced_yf._cache:
cache_info = ticker.get_cache_info()
print(f" Cache status: {cache_info['file_count']} files")
except Exception as e:
print(f"❌ Connection test failed: {e}")
sys.exit(1)
def cmd_benchmark(args):
"""Run performance benchmark"""
print("🏃 Running performance benchmark...")
try:
import time
from yfinance_enhanced import Ticker, download
symbols = ["AAPL", "GOOGL", "MSFT", "TSLA", "NVDA"]
# Test individual tickers
print(f" Testing individual tickers ({len(symbols)} symbols)...")
start_time = time.time()
for symbol in symbols:
ticker = Ticker(symbol)
hist = ticker.history(period="1mo")
print(f" {symbol}: {len(hist)} data points")
individual_time = time.time() - start_time
# Test bulk download
print(f" Testing bulk download...")
start_time = time.time()
data = download(symbols, period="1mo")
bulk_time = time.time() - start_time
print(f"\n📊 Benchmark Results:")
print(f" Individual requests: {individual_time:.2f}s")
print(f" Bulk download: {bulk_time:.2f}s")
print(f" Bulk speedup: {individual_time/bulk_time:.2f}x")
print(f" Data shape: {data.shape}")
except Exception as e:
print(f"❌ Benchmark failed: {e}")
sys.exit(1)
def main():
"""Main CLI entry point"""
parser = argparse.ArgumentParser(
description="YFinance Plus CLI - Cache management and utilities",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
yfp cache info # Show cache information
yfp cache clear # Clear cache (with confirmation)
yfp cache clear --force # Clear cache without confirmation
yfp config show # Show current configuration
yfp test # Test connection to Yahoo Finance
yfp benchmark # Run performance benchmark
"""
)
subparsers = parser.add_subparsers(dest='command', help='Available commands')
# Cache commands
cache_parser = subparsers.add_parser('cache', help='Cache management commands')
cache_subparsers = cache_parser.add_subparsers(dest='cache_command')
# Cache info
info_parser = cache_subparsers.add_parser('info', help='Show cache information')
info_parser.add_argument('--limit', type=int, help='Limit number of files to show')
info_parser.set_defaults(func=cmd_cache_info)
# Cache clear
clear_parser = cache_subparsers.add_parser('clear', help='Clear cache')
clear_parser.add_argument('--force', action='store_true', help='Clear without confirmation')
clear_parser.set_defaults(func=cmd_cache_clear)
# Config commands
config_parser = subparsers.add_parser('config', help='Configuration commands')
config_subparsers = config_parser.add_subparsers(dest='config_command')
# Config show
show_parser = config_subparsers.add_parser('show', help='Show current configuration')
show_parser.set_defaults(func=cmd_config_show)
# Test command
test_parser = subparsers.add_parser('test', help='Test connection to Yahoo Finance')
test_parser.set_defaults(func=cmd_test_connection)
# Benchmark command
benchmark_parser = subparsers.add_parser('benchmark', help='Run performance benchmark')
benchmark_parser.set_defaults(func=cmd_benchmark)
# Parse arguments
args = parser.parse_args()
# Handle no command
if not args.command:
parser.print_help()
return
# Handle cache subcommands
if args.command == 'cache' and not args.cache_command:
cache_parser.print_help()
return
# Handle config subcommands
if args.command == 'config' and not args.config_command:
config_parser.print_help()
return
# Execute command
if hasattr(args, 'func'):
try:
args.func(args)
except KeyboardInterrupt:
print("\n❌ Operation cancelled by user")
sys.exit(1)
else:
parser.print_help()
if __name__ == "__main__":
main()
Write Preview
Loading…
Cancel
Save