diff options
Diffstat (limited to 'src/teradata')
| -rw-r--r-- | src/teradata/__init__.py | 0 | ||||
| -rw-r--r-- | src/teradata/__main__.py | 100 | ||||
| -rw-r--r-- | src/teradata/cache/__init__.py | 3 | ||||
| -rw-r--r-- | src/teradata/cache/file.py | 86 | ||||
| -rw-r--r-- | src/teradata/isolated.py | 8 | ||||
| -rw-r--r-- | src/teradata/picklers.py | 17 | ||||
| -rw-r--r-- | src/teradata/plots.py | 698 | ||||
| -rw-r--r-- | src/teradata/sec.py | 233 | ||||
| -rw-r--r-- | src/teradata/stock.py | 62 | ||||
| -rw-r--r-- | src/teradata/svg.py | 64 |
10 files changed, 1271 insertions, 0 deletions
diff --git a/src/teradata/__init__.py b/src/teradata/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/src/teradata/__init__.py diff --git a/src/teradata/__main__.py b/src/teradata/__main__.py new file mode 100644 index 0000000..d439b35 --- /dev/null +++ b/src/teradata/__main__.py @@ -0,0 +1,100 @@ +from os import getenv + +from edgar import set_identity +lazy from dotenv import load_dotenv +lazy from loguru import logger + +from .picklers import register +lazy from .isolated import isolated +lazy from .plots import ( + plot_all_tickers, + plot_assets_and_roa, + plot_earnings_reaction, + plot_eps, + plot_revenue_and_profitability, + plot_stock_price, + plot_tdc_vs_nasdaq, + plot_valuation_comparison, +) +lazy from .sec import get_financials, get_periodic_filing_dates +lazy from .stock import Tickers, get_stock_data, get_valuation_snapshot +lazy from .svg import split_all_svgs + + +def main() -> None: + # Revenue/Income/Cloud ARR, Assets/ROA, EPS + logger.info("Plotting Revenue/Income/Cloud ARR, Assets/ROA and EPS...") + with isolated() as env: + env.financials = get_financials("TDC", start="2011-01-01") + + plot_revenue_and_profitability(env.financials) + plot_assets_and_roa(env.financials) + plot_eps(env.financials) + + # TDC vs ^IXIC + logger.info("Plotting TDC vs ^IXIC...") + with isolated() as env: + env.tickers = get_stock_data( + ["TDC", "^IXIC"], + start="2007-10-01", + ) + env.tickers = Tickers(env.tickers["Close"].dropna()) + + plot_tdc_vs_nasdaq(env.tickers.cumulative_returns) + + # All tickers + logger.info("Plotting all tickers...") + with isolated() as env: + env.tickers = get_stock_data( + ["TDC", "^IXIC", "^NDX", "IYW", "THNQ", "SNOW"], + start="2007-10-01", # 2023-01-01 + ) + env.tickers = Tickers(env.tickers["Close"].dropna()) + + # plot_tdc_vs_nasdaq(env.tickers.cumulative_returns) + plot_all_tickers(env.tickers.normalized) + + # TDC Ticker since IPO + logger.info("Plotting TDC stock market price since IPO...") + with isolated() as env: + env.tickers = get_stock_data(["TDC"], start="2007-10-01").xs( + "TDC", level=1, axis=1 + ) + + plot_stock_price( + env.tickers, ticker="TDC", sma_windows=(50, 200), show_open=False + ) + + # Market Reaction since SEC reform (2011) + logger.info("Plotting TDC market reaction since SEC reform...") + with isolated() as env: + env.tickers = get_stock_data(["TDC"], start="2011-01-01").xs( + "TDC", level=1, axis=1 + ) + env.filings = get_periodic_filing_dates("TDC", start="2011-01-01") + + plot_earnings_reaction( + env.tickers, env.filings, ticker="TDC", reaction_window_days=3 + ) + + logger.info("Plotting TDC vs. SNOW") + with isolated() as env: + env.tdc = get_valuation_snapshot("TDC") + env.snow = get_valuation_snapshot("SNOW") + + plot_valuation_comparison(env.tdc, env.snow) + + logger.info("Splitting all SVGs into a data and outline layer...") + split_all_svgs() + + +if __name__ == "__main__": + load_dotenv() + register() + + identity = getenv("SEC_EDGAR_IDENTITY") + if identity is None: + raise ValueError("Environment variable 'SEC_EDGAR_IDENTITY' is not set") + set_identity(identity) + + main() diff --git a/src/teradata/cache/__init__.py b/src/teradata/cache/__init__.py new file mode 100644 index 0000000..feb1eb1 --- /dev/null +++ b/src/teradata/cache/__init__.py @@ -0,0 +1,3 @@ +from .file import persistent + +__all__ = ["persistent"] diff --git a/src/teradata/cache/file.py b/src/teradata/cache/file.py new file mode 100644 index 0000000..ebb9bda --- /dev/null +++ b/src/teradata/cache/file.py @@ -0,0 +1,86 @@ +lazy import ast +lazy from collections.abc import Callable +lazy from functools import wraps +lazy from inspect import getsource, signature, unwrap +lazy from pathlib import Path +lazy from sys import version_info +lazy from textwrap import dedent +lazy from types import FunctionType +lazy from uuid import uuid4 + +lazy import dill +lazy from blake3 import blake3 +lazy from loguru import logger + +BASE_FOLDER = Path( + f"./.cache/{version_info.major}.{version_info.minor}.{version_info.micro}/" +) + + +def _function_hash[**P, R](fn: Callable[P, R]) -> str: + fn = unwrap(fn) + + if not isinstance(fn, FunctionType): + raise ValueError(f"fn {fn} is not an instance of types.FunctionType") + + source = getsource(fn) + parsed = ast.parse(dedent(source)) + code = ast.dump(parsed) + sig = signature(fn) + name: str = fn.__qualname__ + + hasher = blake3() + hasher.update(name.encode("utf-8")) + hasher.update(sig.format().encode("utf-8")) + hasher.update(code.encode("utf-8")) + + return hasher.hexdigest() + + +def _parameter_hash[**P, R]( + fn: Callable[P, R], *args: P.args, **kwargs: P.kwargs +) -> str: + fn = unwrap(fn) + sig = signature(fn) + + bound = sig.bind(*args, **kwargs) + bound.apply_defaults() + + serialized = dill.dumps(bound.arguments) + hasher = blake3(serialized) + + return hasher.hexdigest() + + +def persistent[**P, R](func: Callable[P, R]) -> Callable[P, R]: + fn_hash = _function_hash(func) + + folder = BASE_FOLDER / fn_hash + folder.mkdir(parents=True, exist_ok=True) + + @wraps(func) + def wrapper(*args: P.args, **kwargs: P.kwargs) -> R: + param_hash = _parameter_hash(func, *args, **kwargs) + + path = folder / f"{param_hash}.dill" + + if not path.exists(): + logger.opt(colors=True).debug( + f"<white>[<cyan>CACHE</cyan>] [<red>MISS</red>] {func.__qualname__}({args}, {kwargs})</white>" + ) + result = func(*args, **kwargs) + payload = dill.dumps(result) + + temp_path = path.with_suffix(f".tmp.{uuid4().hex}") + temp_path.write_bytes(payload) + temp_path.replace(path) + else: + logger.opt(colors=True).debug( + f"<white>[<cyan>CACHE</cyan>] [<green>HIT</green>] {func.__qualname__}({args}, {kwargs})</white>" + ) + payload = path.read_bytes() + result = dill.loads(payload) + + return result + + return wrapper diff --git a/src/teradata/isolated.py b/src/teradata/isolated.py new file mode 100644 index 0000000..3ca28cb --- /dev/null +++ b/src/teradata/isolated.py @@ -0,0 +1,8 @@ +lazy from collections.abc import Generator +lazy from contextlib import contextmanager +lazy from types import SimpleNamespace + + +@contextmanager +def isolated() -> Generator[SimpleNamespace, None, None]: + yield SimpleNamespace() diff --git a/src/teradata/picklers.py b/src/teradata/picklers.py new file mode 100644 index 0000000..bc0562d --- /dev/null +++ b/src/teradata/picklers.py @@ -0,0 +1,17 @@ +lazy import copyreg + +lazy from lxml import html +lazy from lxml.html._element import HtmlElement + + +def _reduce_html_element(elem: HtmlElement): + raw_html = html.tostring(elem) + return _rebuild_html_element, (raw_html,) + + +def _rebuild_html_element(raw_html: str) -> HtmlElement: + return html.fromstring(raw_html) + + +def register() -> None: + copyreg.pickle(HtmlElement, _reduce_html_element) diff --git a/src/teradata/plots.py b/src/teradata/plots.py new file mode 100644 index 0000000..6e1aa38 --- /dev/null +++ b/src/teradata/plots.py @@ -0,0 +1,698 @@ +lazy from pathlib import Path +lazy from threading import Lock + +lazy import pandas as pd +lazy import plotly.express as px +lazy import plotly.graph_objects as go +lazy from plotly.subplots import make_subplots + +lazy from .stock import ValuationSnapshot + +BG_COLOR = "#fafafa" +GRID_COLOR = "#e2e8f0" + +BASE_LAYOUT = dict( + paper_bgcolor=BG_COLOR, + plot_bgcolor=BG_COLOR, + font=dict(color="#1C1917", family="Inter, Arial, sans-serif"), + hovermode="x unified", + margin=dict(l=60, r=60, t=80, b=50), + legend=dict( + orientation="h", + yanchor="bottom", + y=1.02, + xanchor="right", + x=1, + bgcolor="rgba(250, 250, 250, 0.8)", + ), +) + +EXPORT_DIR = Path("charts") +EXPORT_DIR.mkdir(exist_ok=True) + +_KALEIDO_LOCK = Lock() + + +def export_for_powerpoint( + fig: go.Figure, filename: str, width: int = 1600, height: int = 900 +) -> None: + html_path = EXPORT_DIR / f"{filename}.html" + fig.write_html(str(html_path)) + + png_path = EXPORT_DIR / f"{filename}.png" + fig.write_image(str(png_path), width=width, height=height, scale=2) + + fig.update_layout( + paper_bgcolor="rgba(0,0,0,0)", + plot_bgcolor="rgba(0,0,0,0)", + ) + + svg_path = EXPORT_DIR / f"{filename}.svg" + fig.write_image(str(svg_path), width=width, height=height, scale=2) + + +def plot_revenue_and_profitability( + financials: pd.DataFrame, export: bool = True +) -> go.Figure: + df = financials.copy() + df.index = pd.to_datetime(df.index) + + rev_m = df["revenue"] / 1e6 + cloud_series = df["cloud_arr"].dropna() + cloud_unique = cloud_series[cloud_series != cloud_series.shift(1)] / 1e6 + net_inc_m = df["net_income"] / 1e6 + + fig = make_subplots(specs=[[{"secondary_y": True}]]) + + fig.add_trace( + go.Bar( + x=df.index, + y=rev_m, + name="Quarterly Revenue ($M)", + marker_color="#2b5cff", + opacity=0.85, + ), + secondary_y=False, + ) + fig.add_trace( + go.Scatter( + x=cloud_unique.index, + y=cloud_unique / 4, + name="Cloud Run-Rate (ARR / 4) ($M)", + line=dict(color="#00d2d3", width=3), + mode="lines+markers", + ), + secondary_y=False, + ) + fig.add_trace( + go.Scatter( + x=df.index, + y=net_inc_m, + name="Net Income ($M)", + line=dict(color="#ff9f43", width=3), + mode="lines+markers", + ), + secondary_y=True, + ) + + fig.add_hline(y=0, line_dash="dash", line_color="#cbd5e1", secondary_y=True) + + fig.update_layout( + **BASE_LAYOUT, + title="<b>Teradata (TDC)</b> — Quarterly Revenue, Cloud ARR & Net Income", + barmode="group", + xaxis=dict(title="Quarter End", showgrid=False), + yaxis=dict( + title="Quarterly Revenue / Cloud Run-Rate ($M)", + showgrid=True, + gridcolor=GRID_COLOR, + ), + yaxis2=dict(title="Net Income ($M)", showgrid=False), + ) + + fig.update_xaxes( + range=[ + df.index.min() - pd.Timedelta(days=45), + df.index.max() + pd.Timedelta(days=45), + ] + ) + + if export: + export_for_powerpoint(fig, "01_tdc_revenue_arr_net_income") + # fig.show() + return fig + + +def plot_assets_and_roa(financials: pd.DataFrame, export: bool = True) -> go.Figure: + df = financials.copy() + df.index = pd.to_datetime(df.index) + + assets_b = df["total_assets"] / 1e9 + roa_pct = df["roa_annualized"] * 100 + + fig = make_subplots(specs=[[{"secondary_y": True}]]) + + fig.add_trace( + go.Bar( + x=df.index, + y=assets_b, + name="Total Assets ($B)", + marker_color="#5f27cd", + opacity=0.75, + ), + secondary_y=False, + ) + fig.add_trace( + go.Scatter( + x=df.index, + y=roa_pct, + name="Annualized ROA (%)", + line=dict(color="#10ac84", width=3), + mode="lines+markers", + ), + secondary_y=True, + ) + + fig.add_hline(y=0, line_dash="dash", line_color="#cbd5e1", secondary_y=True) + + fig.update_layout( + **BASE_LAYOUT, + title="<b>Teradata (TDC)</b> — Total Assets & Annualized Return on Assets (ROA)", + xaxis=dict(title="Quarter End", showgrid=False), + yaxis=dict(title="Total Assets ($B)", showgrid=True, gridcolor=GRID_COLOR), + yaxis2=dict(title="Annualized ROA (%)", showgrid=False, ticksuffix="%"), + ) + + fig.update_xaxes( + range=[ + df.index.min() - pd.Timedelta(days=45), + df.index.max() + pd.Timedelta(days=45), + ] + ) + + if export: + export_for_powerpoint(fig, "02_tdc_assets_and_roa") + # fig.show() + return fig + + +def plot_tdc_vs_nasdaq( + cumulative_returns: pd.DataFrame, + benchmark_col: str = "^IXIC", + export: bool = True, +) -> go.Figure: + cum_ret = cumulative_returns.copy() + cum_ret.index = pd.to_datetime(cum_ret.index) + + fig = go.Figure() + fig.add_trace( + go.Scatter( + x=cum_ret.index, + y=cum_ret["TDC"], + name="Teradata (TDC)", + line=dict(color="#ee5253", width=2.5), + ) + ) + fig.add_trace( + go.Scatter( + x=cum_ret.index, + y=cum_ret[benchmark_col], + name=f"Nasdaq Composite ({benchmark_col})", + line=dict(color="#2b5cff", width=2, dash="dash"), + ) + ) + + fig.add_hline(y=0, line_dash="solid", line_color="#718096", opacity=0.6) + + fig.update_layout( + **BASE_LAYOUT, + title="<b>Teradata (TDC) vs. Nasdaq Composite</b> — Cumulative Total Return", + legend_title_text="Benchmark", + yaxis=dict( + title="Cumulative Return (%)", + ticksuffix="%", + showgrid=True, + gridcolor=GRID_COLOR, + ), + xaxis=dict(title="Date", showgrid=False), + ) + + if export: + export_for_powerpoint(fig, "03_tdc_vs_nasdaq_performance") + # fig.show() + return fig + + +def plot_all_tickers(normalized_prices: pd.DataFrame, export: bool = True) -> go.Figure: + df = normalized_prices.copy() + df.index = pd.to_datetime(df.index) + + fig = px.line( + df, + x=df.index, + y=df.columns, + color_discrete_map={ + "TDC": "#dc2626", + "^IXIC": "#60a5fa", + "^NDX": "#2563eb", + "IYW": "#16a34a", + "THNQ": "#d97706", + "SNOW": "#0891b2", + }, + ) + + fig.add_hline( + y=100, + line_dash="dash", + line_color="#94a3b8", + opacity=0.8, + annotation_text="Base 100", + annotation_position="bottom right", + ) + + fig.update_traces(line=dict(width=1.5)) + fig.update_traces(selector=dict(name="TDC"), line=dict(width=3.2)) + + fig.update_layout( + **BASE_LAYOUT, + title="<b>Peer Comparison</b> — Normalized Price Performance (Base = 100)", + legend_title_text="Ticker", + xaxis=dict(title="Date", showgrid=False), + yaxis=dict( + title="Normalized Performance (Base = 100)", + showgrid=True, + gridcolor=GRID_COLOR, + ), + ) + + if export: + export_for_powerpoint(fig, "04_all_peers_normalized_base100") + # fig.show() + return fig + + +def plot_eps(financials: pd.DataFrame, export: bool = True) -> go.Figure: + df = financials.copy() + df.index = pd.to_datetime(df.index) + + fig = go.Figure() + + fig.add_trace( + go.Bar( + x=df.index, + y=df["eps_basic"], + name="Basic EPS ($)", + marker_color="#2563eb", + opacity=0.85, + ) + ) + + fig.add_trace( + go.Bar( + x=df.index, + y=df["eps_diluted"], + name="Diluted EPS ($)", + marker_color="#06b6d4", + opacity=0.85, + ) + ) + + fig.add_hline(y=0, line_dash="solid", line_color="#94a3b8", opacity=0.8) + + fig.update_layout( + **BASE_LAYOUT, + title="<b>Teradata (TDC)</b> — Earnings Per Share (Basic vs. Diluted)", + legend_title_text="Metric", + barmode="group", + xaxis=dict(title="Quarter End", showgrid=False), + yaxis=dict( + title="Earnings Per Share ($)", + tickprefix="$", + showgrid=True, + gridcolor=GRID_COLOR, + ), + ) + + fig.update_xaxes( + range=[ + df.index.min() - pd.Timedelta(days=45), + df.index.max() + pd.Timedelta(days=45), + ] + ) + + if export: + export_for_powerpoint(fig, "05_tdc_eps_basic_vs_diluted") + # fig.show() + return fig + + +def plot_stock_price( + price_data: pd.DataFrame, + ticker: str = "TDC", + sma_windows: tuple[int, ...] = (20, 50), + show_close: bool = True, + show_open: bool = True, + export: bool = True, +) -> go.Figure: + df = price_data.copy() + df.index = pd.to_datetime(df.index) + + df.columns = [str(col).strip().lower() for col in df.columns] + + if "close" not in df.columns or "open" not in df.columns: + raise ValueError("DataFrame must contain 'Open' and 'Close' columns.") + + fig = go.Figure() + + if show_close: + fig.add_trace( + go.Scatter( + x=df.index, + y=df["close"], + name="Close ($)", + line=dict(color="#1e293b", width=2.5), + mode="lines", + ) + ) + + if show_open: + fig.add_trace( + go.Scatter( + x=df.index, + y=df["open"], + name="Open ($)", + line=dict(color="#94a3b8", width=1.5, dash="dot"), + mode="lines", + ) + ) + + sma_colors = ["#2563eb", "#f59e0b", "#10b981", "#8b5cf6"] + for i, window in enumerate(sma_windows): + sma_col = f"sma_{window}" + if sma_col not in df.columns: + df[sma_col] = df["close"].rolling(window=window).mean() + + color = sma_colors[i % len(sma_colors)] + fig.add_trace( + go.Scatter( + x=df.index, + y=df[sma_col], + name=f"{window}-Day SMA ($)", + line=dict(color=color, width=1.8), + mode="lines", + ) + ) + + fig.update_layout( + **BASE_LAYOUT, + title=f"<b>{ticker.upper()}</b> — Daily Stock Price & Moving Averages", + legend_title_text="Series", + xaxis=dict( + title="Date", + showgrid=False, + rangebreaks=[dict(bounds=["sat", "mon"])], + ), + yaxis=dict( + title="Share Price ($)", + tickprefix="$", + showgrid=True, + gridcolor=GRID_COLOR, + ), + ) + + if export: + export_for_powerpoint(fig, f"06_{ticker.lower()}_price_sma") + + # fig.show() + return fig + + +def plot_earnings_reaction( + price_data: pd.DataFrame, + earnings_dates: pd.DataFrame | pd.Series | list, + ticker: str = "TDC", + reaction_window_days: int = 1, + export: bool = True, +) -> go.Figure: + df = price_data.copy() + df.index = pd.to_datetime(df.index) + df.columns = [str(col).strip().lower() for col in df.columns] + + if "close" not in df.columns: + raise ValueError("DataFrame must contain a 'Close' column.") + + df = df.sort_index() + + earnings_labels: dict[pd.Timestamp, str] = {} + if isinstance(earnings_dates, pd.DataFrame): + e_df = earnings_dates.copy() + date_col = next( + (c for c in e_df.columns if "date" in str(c).lower()), e_df.index.name + ) + if date_col and date_col in e_df.columns: + parsed_dates = pd.to_datetime(e_df[date_col]) + else: + parsed_dates = pd.to_datetime(e_df.index) + + label_col = next( + ( + c + for c in e_df.columns + if str(c).lower() in ["quarter", "period", "label", "form"] + ), + None, + ) + if label_col: + for dt, lbl in zip(parsed_dates, e_df[label_col]): + earnings_labels[pd.to_datetime(dt).normalize()] = str(lbl) + raw_dates = parsed_dates + elif isinstance(earnings_dates, pd.Series): + raw_dates = pd.to_datetime(earnings_dates.values) + else: + raw_dates = pd.to_datetime(list(earnings_dates)) + + raw_dates = sorted(pd.DatetimeIndex(raw_dates).dropna().normalize().unique()) + + reactions = [] + for event_date in raw_dates: + trading_days_on_or_after = df.index[df.index >= event_date] + if len(trading_days_on_or_after) == 0: + continue + event_idx = trading_days_on_or_after[0] + event_loc = df.index.get_loc(event_idx) + + base_loc = event_loc - 1 if event_loc > 0 else event_loc + target_loc = min(event_loc + reaction_window_days - 1, len(df) - 1) + + base_price = df["close"].iloc[base_loc] + react_price = df["close"].iloc[target_loc] + pct_change = ((react_price - base_price) / base_price) * 100 + + label = earnings_labels.get(event_date, "Earnings Call") + + reactions.append( + { + "date": df.index[target_loc], + "price": react_price, + "pct_change": pct_change, + "label": label, + } + ) + + react_df = pd.DataFrame(reactions) + + fig = go.Figure() + + fig.add_trace( + go.Scatter( + x=df.index, + y=df["close"], + name=f"{ticker.upper()} Close ($)", + line=dict(color="#334155", width=2), + mode="lines", + hoverinfo="skip", + ) + ) + + if not react_df.empty: + pos_df = react_df[react_df["pct_change"] >= 0] + neg_df = react_df[react_df["pct_change"] < 0] + + pos_sizes = (pos_df["pct_change"].abs() * 0.7 + 8).clip(8, 20) + neg_sizes = (neg_df["pct_change"].abs() * 0.7 + 8).clip(8, 20) + + pos_y = pos_df["price"] * 0.965 + neg_y = neg_df["price"] * 1.035 + + fig.add_trace( + go.Scatter( + x=pos_df["date"], + y=pos_y, + name="Positive Reaction (▲)", + mode="markers", + marker=dict( + symbol="triangle-up", + size=pos_sizes, + color="#10b981", + line=dict(width=1, color="#065f46"), + ), + customdata=list( + zip(pos_df["pct_change"], pos_df["label"], pos_df["price"]) + ), + hovertemplate=( + "<b>%{customdata[1]}</b><br>" + "Date: %{x|%Y-%m-%d}<br>" + "Reaction: +%{customdata[0]:.2f}%<br>" + "Close: $%{customdata[2]:.2f}<extra></extra>" + ), + ) + ) + + fig.add_trace( + go.Scatter( + x=neg_df["date"], + y=neg_y, + name="Negative Reaction (▼)", + mode="markers", + marker=dict( + symbol="triangle-down", + size=neg_sizes, + color="#ef4444", + line=dict(width=1, color="#7f1d1d"), + ), + customdata=list( + zip(neg_df["pct_change"], neg_df["label"], neg_df["price"]) + ), + hovertemplate=( + "<b>%{customdata[1]}</b><br>" + "Date: %{x|%Y-%m-%d}<br>" + "Reaction: %{customdata[0]:.2f}%<br>" + "Close: $%{customdata[2]:.2f}<extra></extra>" + ), + ) + ) + + fig.update_layout( + **BASE_LAYOUT, + title=f"<b>{ticker.upper()}</b> — Stock Price Reaction to Earnings Announcements", + legend_title_text="Reaction", + xaxis=dict( + title="Date", + showgrid=False, + rangebreaks=[dict(bounds=["sat", "mon"])], + ), + yaxis=dict( + title="Share Price ($)", + tickprefix="$", + showgrid=True, + gridcolor=GRID_COLOR, + ), + ) + + fig.update_xaxes( + range=[ + df.index.min() - pd.Timedelta(days=45), + df.index.max() + pd.Timedelta(days=45), + ] + ) + + if export: + export_for_powerpoint(fig, f"07_{ticker.lower()}_earnings_reactions") + + # fig.show() + return fig + + +def plot_valuation_comparison( + tdc: ValuationSnapshot, snow: ValuationSnapshot, export: bool = True +) -> go.Figure: + categories = [tdc.ticker, snow.ticker] + bar_colors = ["#2b5cff", "#00d2d3"] + + fig = make_subplots( + rows=1, + cols=3, + subplot_titles=( + "<b>EV / Sales (TTM)</b>", + "<b>YoY Revenue Growth (%)</b>", + "<b>EV / FCF (TTM)</b>", + ), + horizontal_spacing=0.08, + ) + + ev_sales_vals = [tdc.ev_to_sales or 0, snow.ev_to_sales or 0] + ev_sales_text = [ + f"{tdc.ev_to_sales:.1f}x" if tdc.ev_to_sales is not None else "N/A", + f"{snow.ev_to_sales:.1f}x" if snow.ev_to_sales is not None else "N/A", + ] + fig.add_trace( + go.Bar( + x=categories, + y=ev_sales_vals, + text=ev_sales_text, + textposition="outside", + cliponaxis=False, + marker_color=bar_colors, + opacity=0.85, + showlegend=False, + ), + row=1, + col=1, + ) + + rev_growth_vals = [tdc.revenue_growth_yoy or 0, snow.revenue_growth_yoy or 0] + rev_growth_text = [ + ( + f"{tdc.revenue_growth_yoy:+.1f}%" + if tdc.revenue_growth_yoy is not None + else "N/A" + ), + ( + f"{snow.revenue_growth_yoy:+.1f}%" + if snow.revenue_growth_yoy is not None + else "N/A" + ), + ] + fig.add_trace( + go.Bar( + x=categories, + y=rev_growth_vals, + text=rev_growth_text, + textposition="outside", + cliponaxis=False, + marker_color=bar_colors, + opacity=0.85, + showlegend=False, + ), + row=1, + col=2, + ) + + ev_fcf_vals = [tdc.ev_to_fcf or 0, snow.ev_to_fcf or 0] + ev_fcf_text = [ + f"{tdc.ev_to_fcf:.1f}x" if tdc.ev_to_fcf is not None else "N/A", + f"{snow.ev_to_fcf:.1f}x" if snow.ev_to_fcf is not None else "N/A", + ] + fig.add_trace( + go.Bar( + x=categories, + y=ev_fcf_vals, + text=ev_fcf_text, + textposition="outside", + cliponaxis=False, + marker_color=bar_colors, + opacity=0.85, + showlegend=False, + ), + row=1, + col=3, + ) + + fig.update_layout(**BASE_LAYOUT) + fig.update_layout( + title="<b>Valuation Disconnect</b> — Teradata (TDC) vs. Snowflake (SNOW)", + showlegend=False, + hovermode="closest", + ) + + for annotation in fig["layout"]["annotations"]: + annotation["font"] = dict( + family="Inter, Arial, sans-serif", + size=14, + color="#1C1917", + ) + + fig.update_xaxes(showgrid=False, tickfont=dict(size=12)) + fig.update_yaxes( + showgrid=True, + gridcolor=GRID_COLOR, + zeroline=True, + zerolinecolor="#cbd5e1", + zerolinewidth=1, + ) + + if export: + export_for_powerpoint(fig, "08_valuation_comparison_tdc_vs_snow") + + # fig.show() + return fig diff --git a/src/teradata/sec.py b/src/teradata/sec.py new file mode 100644 index 0000000..fbba073 --- /dev/null +++ b/src/teradata/sec.py @@ -0,0 +1,233 @@ +""" +This is by no means a complete implementation of on-disk caching. +It however serves its purpose for my specific need. +""" + +lazy import re + +lazy import pandas as pd +lazy from bs4 import BeautifulSoup +lazy from edgar import Company + +lazy from .cache import persistent + +CONCEPTS = { + "revenue": [ + "RevenueFromContractWithCustomerExcludingAssessedTax", + "SalesRevenueNet", + "Revenues", + ], + "net_income": [ + "NetIncomeLoss", + "ProfitLoss", + "NetIncomeLossAvailableToCommonStockholdersBasic", + ], + "eps_basic": ["EarningsPerShareBasic"], + "eps_diluted": ["EarningsPerShareDiluted"], + "total_assets": ["Assets"], +} + +MONEY = r"\$\s?(\d[\d,]*\.?\d*)\s*(million|billion|M|B)\b" +ARR_RE = [ + re.compile(rf"Cloud\s+ARR[^$.]{{0,150}}?{MONEY}", re.I), + re.compile(rf"{MONEY}\s+(?:of|in)\s+Cloud\s+ARR", re.I), +] + + +def clean_facts(sub: pd.DataFrame) -> pd.DataFrame: + df = sub.dropna(subset=["period_end", "numeric_value"]).copy() + for dim_col in ["dimensions", "segments", "dimension"]: + if dim_col in df.columns: + df = df[df[dim_col].isna() | (df[dim_col] == "") | (df[dim_col] == "{}")] + return df.sort_values("filing_date") + + +def parse_flow_concept(sub: pd.DataFrame) -> pd.Series: + sub = clean_facts(sub) + if sub.empty: + return pd.Series(dtype=float) + + sub = sub.dropna(subset=["period_start"]).copy() + sub["days"] = (sub["period_end"] - sub["period_start"]).dt.days + + sub = sub[sub["days"].between(70, 380)] + + sub = sub.sort_values("filing_date").drop_duplicates( + ["period_start", "period_end"], keep="last" + ) + + q_discrete = ( + sub[sub["days"].between(75, 105)] + .set_index("period_end")["numeric_value"] + .to_dict() + ) + + ytd_6m = sub[sub["days"].between(165, 200)] + for _, r in ytd_6m.iterrows(): + pend = r["period_end"] + if pend not in q_discrete: + q1_match = [ + v + for k, v in q_discrete.items() + if 70 <= (pend - k).days <= 105 and k > r["period_start"] + ] + if q1_match: + q_discrete[pend] = r["numeric_value"] - q1_match[0] + + ytd_9m = sub[sub["days"].between(255, 290)] + for _, r in ytd_9m.iterrows(): + pend = r["period_end"] + if pend not in q_discrete: + in_range_q = [ + v for k, v in q_discrete.items() if r["period_start"] < k < pend + ] + if len(in_range_q) >= 2: + q_discrete[pend] = r["numeric_value"] - sum(in_range_q[-2:]) + + fy_12m = sub[sub["days"].between(350, 380)] + for _, r in fy_12m.iterrows(): + pend = r["period_end"] + if pend not in q_discrete: + inner_q = [v for k, v in q_discrete.items() if r["period_start"] < k < pend] + if len(inner_q) == 3: + q_discrete[pend] = r["numeric_value"] - sum(inner_q) + + res = pd.Series(q_discrete).sort_index() + res.index = pd.to_datetime(res.index) + return res + + +def parse_instant_concept(sub: pd.DataFrame) -> pd.Series: + sub = clean_facts(sub) + if sub.empty: + return pd.Series(dtype=float) + + sub = sub.sort_values(["period_end", "numeric_value"]) + series = sub.drop_duplicates("period_end", keep="last").set_index("period_end")[ + "numeric_value" + ] + series.index = pd.to_datetime(series.index) + + full_idx = pd.date_range( + start=series.index.min(), end=series.index.max(), freq="QE" + ) + series = series.reindex(series.index.union(full_idx)).sort_index() + return series.ffill().loc[full_idx] + + +def extract_revenue(facts: pd.DataFrame) -> pd.Series: + rev_series = pd.Series(dtype=float) + for tag in [ + "RevenueFromContractWithCustomerExcludingAssessedTax", + "SalesRevenueNet", + "Revenues", + ]: + sub = facts[facts["concept"].astype(str).str.endswith(tag)] + if not sub.empty: + parsed = parse_flow_concept(sub) + valid = parsed[parsed > 300e6] + rev_series = rev_series.combine_first(valid) + + prod_sub = facts[facts["concept"].astype(str).str.endswith("SalesRevenueGoodsNet")] + serv_sub = facts[ + facts["concept"].astype(str).str.endswith("SalesRevenueServicesNet") + ] + + if not prod_sub.empty and not serv_sub.empty: + p_rev = parse_flow_concept(prod_sub) + s_rev = parse_flow_concept(serv_sub) + combined_ps = (p_rev + s_rev).dropna() + rev_series = rev_series.combine_first(combined_ps) + + return rev_series.sort_index() + + +def cloud_arr(ticker: str, since: str) -> pd.DataFrame: + rows = {} + for f in sorted( + Company(ticker).get_filings(form="8-K", filing_date=f"{since}:"), + key=lambda f: f.filing_date, + ): + if "2.02" not in str(f.items or ""): + continue + text = " ".join( + re.sub( + r"\s+", " ", BeautifulSoup(a.download(), "html.parser").get_text(" ") + ) + for a in f.attachments + if str(a.document_type).upper().startswith("EX-99") + ) + for pat in ARR_RE: + if m := pat.search(text): + num, unit = m.groups()[-2:] + multiplier = 1e9 if unit[0].lower() == "b" else 1e6 + val = float(num.replace(",", "")) * multiplier + rows[pd.Timestamp(f.filing_date) - pd.offsets.QuarterEnd(1)] = { + "cloud_arr": val, + "filing_date": f.filing_date, + "snippet": text[max(0, m.start() - 40) : m.end() + 60], + } + break + return pd.DataFrame.from_dict(rows, orient="index").sort_index() + + +@persistent +def get_financials(ticker: str = "TDC", start: str = "2011-01-01") -> pd.DataFrame: + facts = Company(ticker).get_facts().to_dataframe(pit_mode=True) + for c in ("period_start", "period_end", "filing_date"): + if c in facts.columns: + facts[c] = pd.to_datetime(facts[c], errors="coerce") + + series = {} + + asset_facts = facts[facts["concept"].astype(str).str.endswith("Assets")] + series["total_assets"] = parse_instant_concept(asset_facts) + + series["revenue"] = extract_revenue(facts) + + for name in ["net_income", "eps_basic", "eps_diluted"]: + concept_candidates = CONCEPTS[name] + combined = pd.Series(dtype=float) + for tag in concept_candidates: + sub = facts[facts["concept"].astype(str).str.endswith(tag)] + if not sub.empty: + combined = combined.combine_first(parse_flow_concept(sub)) + series[name] = combined + + out = pd.DataFrame(series).sort_index() + + out = out.resample("QE").last() + + out["total_assets"] = out["total_assets"].ffill() + out["avg_assets"] = out["total_assets"].rolling(2).mean() + out["roa_quarterly"] = out["net_income"] / out["avg_assets"] + out["roa_annualized"] = out["roa_quarterly"] * 4 + + arr = cloud_arr(ticker, since="2020-01-01") + if not arr.empty: + arr.index = pd.to_datetime(arr.index).to_period("Q").to_timestamp("Q") + out["cloud_arr"] = arr["cloud_arr"] + out["cloud_arr_yoy"] = out["cloud_arr"].pct_change(4) + + df = out[out.index >= start].copy() + df.index.name = "quarter_end" + return df + + +@persistent +def get_periodic_filing_dates( + ticker: str = "TDC", start: str = "2011-01-01" +) -> pd.DataFrame: + company = Company(ticker) + filings = company.get_filings( + form=["10-K", "10-Q"], + filing_date=f"{start}:", + amendments=False, + ) + + df = filings.to_pandas() + df.columns = [c.lower() for c in df.columns] + + df["filing_date"] = pd.to_datetime(df["filing_date"]) + df = df.sort_values("filing_date").reset_index(drop=True) + return df[["filing_date", "form"]] diff --git a/src/teradata/stock.py b/src/teradata/stock.py new file mode 100644 index 0000000..58d34b4 --- /dev/null +++ b/src/teradata/stock.py @@ -0,0 +1,62 @@ +lazy from typing import Self + +from attrs import frozen +lazy import yfinance +lazy from pandas import DataFrame + +from .cache import persistent + + +@frozen +class Tickers: + prices: DataFrame + + @property + def normalized(self: Self) -> DataFrame: + """Indexed to 100 at t_0.""" + return (self.prices / self.prices.iloc[0]) * 100 + + @property + def cumulative_returns(self: Self) -> DataFrame: + """Percentage gain/loss since t_0.""" + return ((self.prices / self.prices.iloc[0]) - 1) * 100 + + +@frozen +class ValuationSnapshot: + ticker: str + ev_to_sales: float | None = None + revenue_growth_yoy: float | None = None + ev_to_fcf: float | None = None + trailing_pe: float | None = None + forward_pe: float | None = None + + +@persistent +def get_stock_data(tickers: list[str], start: str) -> DataFrame: + data = yfinance.download(tickers, start=start) + if data is None: + raise ValueError(f"Yahoo Finance data for {tickers} from {start} is None") + + return data + + +@persistent +def get_valuation_snapshot(ticker: str) -> ValuationSnapshot: + y_ticker = yfinance.Ticker(ticker) + + info = y_ticker.info + ev = info.get("enterpriseValue") + fcf = info.get("freeCashflow") + ev_fcf = (ev / fcf) if (ev and fcf and fcf > 0) else None + + return ValuationSnapshot( + ticker=ticker, + ev_to_sales=info.get("enterpriseToRevenue"), + revenue_growth_yoy=( + info.get("revenueGrowth", 0.0) * 100 + if info.get("revenueGrowth") is not None + else None + ), + ev_to_fcf=ev_fcf, + ) diff --git a/src/teradata/svg.py b/src/teradata/svg.py new file mode 100644 index 0000000..26546c9 --- /dev/null +++ b/src/teradata/svg.py @@ -0,0 +1,64 @@ +from pathlib import Path + +from lxml.etree import Element, XMLParser, fromstring, tostring + +PARSER = XMLParser(remove_blank_text=False) +BASE_DIR = Path("charts") + + +def read_svg(path: Path) -> Element: + raw = path.read_bytes() + + return fromstring(raw, PARSER) + + +def write_svg(path: Path, element: Element) -> None: + raw = tostring(element, encoding="utf-8", pretty_print=False, xml_declaration=True) + + path.write_bytes(raw) + + +def split_plotly_svg(input_path: Path, outline_path: Path, data_path: Path) -> None: + outline_tree, data_tree = read_svg(input_path), read_svg(input_path) + + data_xpath = "//*[local-name()='g' and contains(@class, 'overplot')]" + for element in outline_tree.xpath(data_xpath): + element.attrib["display"] = "none" + + write_svg(outline_path, outline_tree) + + frame_layers = [ + "bglayer", + "gridlayer", + "zerolinelayer", + "axis", + "infolayer", + "shapelayer", + "overaxes-above", + ] + frame_conditions = " or ".join( + [f"contains(@class, '{cls}')" for cls in frame_layers] + ) + frame_xpath = f"//*[local-name()='g' and ({frame_conditions})]" + for element in data_tree.xpath(frame_xpath): + element.attrib["display"] = "none" + + bg_rect_xpath = "//*[local-name()='rect' and (contains(@class, 'bg') or contains(@class, 'plot-bg'))]" + for rect in data_tree.xpath(bg_rect_xpath): + rect.attrib["display"] = "none" + + write_svg(data_path, data_tree) + + +def split_all_svgs() -> None: + paths = list(BASE_DIR.glob("*.svg")) + + for path in paths: + if path.stem.endswith(("_outline", "_data")): + continue + + split_plotly_svg( + path, + path.with_stem(f"{path.stem}_outline"), + path.with_stem(f"{path.stem}_data"), + ) |
