Spaces:
Running
Running
File size: 4,118 Bytes
eb606e1 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 |
import requests
import yfinance as yf
import pandas as pd
from datetime import datetime, timedelta
from typing import List, Dict, Any, Optional, Tuple
from app.config import settings
from app.models.fund_models import FundMeta, NAVData, FundNAVResponse, MarketIndex, MarketIndicesResponse
class MutualFundDataFetcher:
"""Fetch mutual fund data from MFAPI.in and Yahoo Finance for indices"""
def __init__(self):
self.mfapi_base = settings.MFAPI_BASE_URL
def get_all_schemes(self) -> List[Dict[str, Any]]:
"""Fetch all mutual fund schemes"""
try:
response = requests.get(self.mfapi_base)
if response.status_code == 200:
return response.json()
else:
return []
except Exception as e:
print(f"Error fetching schemes: {e}")
return []
def get_fund_nav_history(self, scheme_code: str) -> Tuple[pd.DataFrame, FundMeta]:
"""Fetch NAV history for a specific scheme"""
try:
url = f"{self.mfapi_base}/{scheme_code}"
response = requests.get(url)
if response.status_code == 200:
data = response.json()
nav_data = data.get('data', [])
# Convert to DataFrame
df = pd.DataFrame(nav_data)
if not df.empty:
df['date'] = pd.to_datetime(df['date'], format='%d-%m-%Y')
df['nav'] = pd.to_numeric(df['nav'])
df = df.sort_values('date')
# Create FundMeta object
meta_data = data.get('meta', {})
fund_meta = FundMeta(
fund_house=meta_data.get('fund_house'),
scheme_type=meta_data.get('scheme_type'),
scheme_category=meta_data.get('scheme_category'),
scheme_code=str(meta_data.get('scheme_code', scheme_code))
)
return df, fund_meta
else:
return pd.DataFrame(), FundMeta()
except Exception as e:
print(f"Error fetching NAV data: {e}")
return pd.DataFrame(), FundMeta()
def get_market_indices(self) -> MarketIndicesResponse:
"""Fetch Indian market indices using Yahoo Finance"""
indices = {
'^NSEI': 'Nifty 50',
'^BSESN': 'BSE Sensex',
'^NSEBANK': 'Nifty Bank',
'^CNXIT': 'Nifty IT',
'^NSEMDCP50': 'Nifty Midcap 50',
'NIFTYSMLCAP50.NS': 'Nifty Smallcap 50',
'^CNXPHARMA': 'Nifty Pharma',
'^CNXAUTO': 'Nifty Auto',
'^CNXFMCG': 'Nifty FMCG',
'^CNXENERGY': 'Nifty Energy',
'^CNXREALTY': 'Nifty Realty',
'^NSMIDCP': 'Nifty Next 50',
}
indices_data = []
for symbol, name in indices.items():
try:
ticker = yf.Ticker(symbol)
hist = ticker.history(period="5d")
if not hist.empty:
current_price = hist['Close'].iloc[-1]
previous_close = hist['Close'].iloc[-2] if len(hist) > 1 else current_price
change = current_price - previous_close
change_pct = (change / previous_close) * 100
indices_data.append(MarketIndex(
name=name,
symbol=symbol,
current_price=current_price,
change=change,
change_pct=change_pct
))
except Exception as e:
print(f"Could not fetch {name}: {e}")
return MarketIndicesResponse(
indices=indices_data,
last_updated=datetime.now()
) |