Initial commitHEADmain

1 files changed, 233 insertions, 0 deletions

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"]]