summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
authorMikka Petterson <m.petterson@student.maastrichtuniversity.nl>2026-10-14 07:45:20 +0200
committerMikka Petterson <m.petterson@student.maastrichtuniversity.nl>2026-10-14 07:45:20 +0200
commit0bcc5be35b6a99109bd04925c22890347c9d6e88 (patch)
treefdd88728ad1747268b0a02aba0de08a3af6a52b2 /src
Initial commitHEADmain
Diffstat (limited to 'src')
-rw-r--r--src/teradata/__init__.py0
-rw-r--r--src/teradata/__main__.py100
-rw-r--r--src/teradata/cache/__init__.py3
-rw-r--r--src/teradata/cache/file.py86
-rw-r--r--src/teradata/isolated.py8
-rw-r--r--src/teradata/picklers.py17
-rw-r--r--src/teradata/plots.py698
-rw-r--r--src/teradata/sec.py233
-rw-r--r--src/teradata/stock.py62
-rw-r--r--src/teradata/svg.py64
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"),
+ )