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
102
103
104
105
106
107
108
109
|
"""yfinance wrapper — price history, financial statements, company info."""
import yfinance as yf
import pandas as pd
import streamlit as st
@st.cache_data(ttl=60)
def search_tickers(query: str) -> list[dict]:
"""Search for tickers by company name or symbol. Returns list of {symbol, name, exchange}."""
if not query or len(query.strip()) < 2:
return []
try:
results = yf.Search(query.strip(), max_results=8).quotes
out = []
for r in results:
symbol = r.get("symbol", "")
name = r.get("longname") or r.get("shortname") or symbol
exchange = r.get("exchange") or r.get("exchDisp", "")
if symbol:
out.append({"symbol": symbol, "name": name, "exchange": exchange})
return out
except Exception:
return []
@st.cache_data(ttl=300)
def get_company_info(ticker: str) -> dict:
"""Return company info dict from yfinance."""
t = yf.Ticker(ticker.upper())
info = t.info or {}
return info
@st.cache_data(ttl=300)
def get_price_history(ticker: str, period: str = "1y") -> pd.DataFrame:
"""Return OHLCV price history."""
t = yf.Ticker(ticker.upper())
df = t.history(period=period)
df.index = pd.to_datetime(df.index)
return df
@st.cache_data(ttl=3600)
def get_income_statement(ticker: str, quarterly: bool = False) -> pd.DataFrame:
t = yf.Ticker(ticker.upper())
df = t.quarterly_income_stmt if quarterly else t.income_stmt
return df if df is not None else pd.DataFrame()
@st.cache_data(ttl=3600)
def get_balance_sheet(ticker: str, quarterly: bool = False) -> pd.DataFrame:
t = yf.Ticker(ticker.upper())
df = t.quarterly_balance_sheet if quarterly else t.balance_sheet
return df if df is not None else pd.DataFrame()
@st.cache_data(ttl=3600)
def get_cash_flow(ticker: str, quarterly: bool = False) -> pd.DataFrame:
t = yf.Ticker(ticker.upper())
df = t.quarterly_cashflow if quarterly else t.cashflow
return df if df is not None else pd.DataFrame()
@st.cache_data(ttl=300)
def get_market_indices() -> dict:
"""Return latest price + day change % for major indices."""
symbols = {
"S&P 500": "^GSPC",
"NASDAQ": "^IXIC",
"DOW": "^DJI",
"VIX": "^VIX",
}
result = {}
for name, sym in symbols.items():
try:
t = yf.Ticker(sym)
hist = t.history(period="2d")
if len(hist) >= 2:
prev_close = hist["Close"].iloc[-2]
last = hist["Close"].iloc[-1]
pct_change = (last - prev_close) / prev_close
elif len(hist) == 1:
last = hist["Close"].iloc[-1]
pct_change = 0.0
else:
result[name] = {"price": None, "change_pct": None}
continue
result[name] = {"price": float(last), "change_pct": float(pct_change)}
except Exception:
result[name] = {"price": None, "change_pct": None}
return result
@st.cache_data(ttl=3600)
def get_free_cash_flow_series(ticker: str) -> pd.Series:
"""Return annual Free Cash Flow series (most recent first)."""
t = yf.Ticker(ticker.upper())
cf = t.cashflow
if cf is None or cf.empty:
return pd.Series(dtype=float)
if "Free Cash Flow" in cf.index:
return cf.loc["Free Cash Flow"].dropna()
# Compute from operating CF - capex
try:
op = cf.loc["Operating Cash Flow"]
capex = cf.loc["Capital Expenditure"]
return (op + capex).dropna()
except KeyError:
return pd.Series(dtype=float)
|
