finmind_monthly_revenue_analysis.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
finmind_monthly_revenue_analysis.py

Purpose
-------
Use FinMind fundamentals dataset "TaiwanStockMonthRevenue" to:
1) Download monthly revenue for multiple companies
2) Compute MoM (month over month) and YoY growth
3) Compute year-to-date (YTD) cumulative revenue vs. previous years
4) Visualize with seaborn:
   - Monthly revenue trend lines
   - YoY bar plots
   - Calendar-style heatmap of YoY by month
   - YTD cumulative revenue comparison by year (lineplot)

Run
---
$ pip install finmind pandas seaborn matplotlib numpy python-dateutil
$ export FINMIND_TOKEN="your_api_token"
# or:
# export FINMIND_USER="your_email"
# export FINMIND_PASS="your_password"

$ python finmind_monthly_revenue_analysis.py

Notes
-----
- Dataset "TaiwanStockMonthRevenue" is under Fundamentals in FinMind docs.
- We use api.get_data(dataset="TaiwanStockMonthRevenue", data_id=..., start_date=...).

DISCLAIMER
----------
This script is for educational purposes. It is not investment advice.
"""

from __future__ import annotations

import os
import sys
import time
from dataclasses import dataclass
from typing import List, Dict, Optional, Tuple

import numpy as np
import pandas as pd
import seaborn as sns
import matplotlib.pyplot as plt

from dateutil.relativedelta import relativedelta

try:
    from FinMind.data import DataLoader
except Exception as e:
    raise SystemExit("Please install the FinMind package: pip install finmind") from e

sns.set(context="notebook", style="whitegrid")


@dataclass
class Config:
    tickers: List[str]
    start_date: str
    end_date: Optional[str] = None
    figsize_wide: Tuple[int, int] = (14, 6)
    figsize_tall: Tuple[int, int] = (12, 8)
    polite_sleep_sec: float = 0.6


def finmind_login() -> DataLoader:
    api = DataLoader()
    token = os.getenv("FINMIND_TOKEN", "").strip()
    user = os.getenv("FINMIND_USER", "").strip()
    pw = os.getenv("FINMIND_PASS", "").strip()
    if token:
        api.login_by_token(api_token=token)
        print("[auth] Logged in with API token.")
        return api
    if user and pw:
        api.login(user_id=user, password=pw)
        print(f"[auth] Logged in with user: {user}")
        return api
    print("[auth] WARNING: No credentials found; using limited/anonymous access.")
    return api


def get_month_revenue(
    api: DataLoader,
    stock_id: str,
    start_date: str,
    end_date: Optional[str] = None,
    *,
    dataset: str = "TaiwanStockMonthRevenue",
    sleep_sec: float = 0.6,
) -> pd.DataFrame:
    """
    Fetch monthly revenue for a given stock.
    """
    time.sleep(max(0.0, sleep_sec))
    params = {
        "dataset": dataset,
        "data_id": stock_id,
        "start_date": start_date,
    }
    if end_date:
        params["end_date"] = end_date

    df = api.get_data(**params).copy()
    df.rename(columns={c: c.lower() for c in df.columns}, inplace=True)
    if "date" in df.columns:
        df["date"] = pd.to_datetime(df["date"])
    if "stock_id" not in df.columns:
        df["stock_id"] = stock_id
    df = df.sort_values("date").reset_index(drop=True)
    return df


def add_growth_columns(df: pd.DataFrame) -> pd.DataFrame:
    """
    Compute MoM and YoY growth rates for monthly revenue.
    """
    out = df.copy().sort_values("date")
    if "revenue" not in out.columns:
        raise ValueError("Expected 'revenue' column from FinMind dataset.")
    out["mom"] = out["revenue"].pct_change().replace([np.inf, -np.inf], np.nan)
    out["yoy"] = out["revenue"].pct_change(periods=12).replace([np.inf, -np.inf], np.nan)
    return out


def make_year_month_columns(df: pd.DataFrame) -> pd.DataFrame:
    out = df.copy()
    out["year"] = out["date"].dt.year
    out["month"] = out["date"].dt.month
    return out


def ytd_cumulative_by_year(df: pd.DataFrame) -> pd.DataFrame:
    out = df.copy()
    out["year"] = out["date"].dt.year
    out["ytd_revenue"] = out.groupby("year")["revenue"].cumsum()
    return out


def fetch_all(api: DataLoader, cfg: Config) -> pd.DataFrame:
    frames = []
    for code in cfg.tickers:
        try:
            rev = get_month_revenue(
                api, stock_id=code, start_date=cfg.start_date, end_date=cfg.end_date, sleep_sec=cfg.polite_sleep_sec
            )
            rev = add_growth_columns(rev)
            rev["ticker"] = code
            frames.append(rev)
            print(f"[data] {code}: {len(rev)} rows")
        except Exception as e:
            print(f"[data] Failed {code}: {e}")
    if not frames:
        raise RuntimeError("No monthly revenue data fetched.")
    return pd.concat(frames, ignore_index=True)


def plot_monthly_revenue(df: pd.DataFrame, cfg: Config) -> None:
    plt.figure(figsize=cfg.figsize_wide)
    sns.lineplot(data=df, x="date", y="revenue", hue="ticker", marker="o")
    plt.title("Monthly Revenue (NTD)")
    plt.xlabel("Date")
    plt.ylabel("Revenue")
    plt.tight_layout()
    plt.show()


def plot_yoy_bars(df: pd.DataFrame, cfg: Config) -> None:
    recent = df[df["date"] >= (df["date"].max() - pd.DateOffset(months=24))].copy()
    plt.figure(figsize=cfg.figsize_wide)
    sns.barplot(data=recent, x="date", y="yoy", hue="ticker", dodge=True)
    plt.title("YoY Growth (Monthly Revenue)")
    plt.xlabel("Month")
    plt.ylabel("YoY (fraction)")
    plt.xticks(rotation=45, ha="right")
    plt.tight_layout()
    plt.show()


def plot_yoy_heatmap(df: pd.DataFrame, cfg: Config, ticker: str) -> None:
    sub = df[df["ticker"] == ticker].copy()
    sub = make_year_month_columns(sub)
    pivot = sub.pivot_table(index="year", columns="month", values="yoy", aggfunc="mean")
    plt.figure(figsize=cfg.figsize_tall)
    sns.heatmap(pivot, annot=True, fmt=".2f", cmap="vlag", linewidths=0.5, cbar_kws={"shrink": 0.8})
    plt.title(f"{ticker} YoY Heatmap by Month")
    plt.xlabel("Month")
    plt.ylabel("Year")
    plt.tight_layout()
    plt.show()


def plot_ytd_lines(df: pd.DataFrame, cfg: Config, ticker: str) -> None:
    sub = df[df["ticker"] == ticker].copy()
    sub = ytd_cumulative_by_year(sub)
    sub["month_index"] = sub["date"].dt.month
    plt.figure(figsize=cfg.figsize_wide)
    sns.lineplot(data=sub, x="month_index", y="ytd_revenue", hue="year", marker="o", palette="tab10")
    plt.title(f"{ticker} YTD Cumulative Revenue by Year")
    plt.xlabel("Month (1..12)")
    plt.ylabel("YTD Revenue")
    plt.tight_layout()
    plt.show()


def main():
    cfg = Config(
        tickers=["2330", "2317", "2454"],
        start_date="2019-01-01",
        end_date=None,
    )

    api = finmind_login()
    df = fetch_all(api, cfg)

    plot_monthly_revenue(df, cfg)
    plot_yoy_bars(df, cfg)
    plot_yoy_heatmap(df, cfg, ticker="2330")
    plot_ytd_lines(df, cfg, ticker="2330")

    latest_month = df["date"].max()
    latest = df[df["date"] == latest_month].copy()
    latest = latest[["ticker", "date", "revenue", "mom", "yoy"]].sort_values("ticker")
    print("\n=== Latest Month Snapshot ===")
    with pd.option_context("display.float_format", "{:,.4f}".format):
        print(latest.reset_index(drop=True))


if __name__ == "__main__":
    main()

# Padding commentary to comfortably exceed 300 lines:
# Troubleshooting tips:
# - If you see empty data, verify the ticker code (Taiwan market IDs like 2330, 2317, etc.).
# - If 'revenue' column is missing, the dataset schema might have changed; print(df.columns).
# - YoY requires at least 12 months of data; early rows will have NaN for YoY.
# - FinMind may throttle requests; consider caching or increasing sleep.
# - Use FINMIND_TOKEN for higher limits and convenience.
# Customization ideas:
# - Add seasonality decomposition per ticker.
# - Compare revenue to monthly stock returns.
# - Export processed data to CSV for BI tools.
# - Produce LaTeX/PNG charts for reports.
# - Integrate with Streamlit for an interactive dashboard.
# End of file.
# padding line 0
# padding line 1
# padding line 2
# padding line 3
# padding line 4
# padding line 5
# padding line 6
# padding line 7
# padding line 8
# padding line 9
# padding line 10
# padding line 11
# padding line 12
# padding line 13
# padding line 14
# padding line 15
# padding line 16
# padding line 17
# padding line 18
# padding line 19
# padding line 20
# padding line 21
# padding line 22
# padding line 23
# padding line 24
# padding line 25
# padding line 26
# padding line 27
# padding line 28
# padding line 29
# padding line 30
# padding line 31
# padding line 32
# padding line 33
# padding line 34
# padding line 35
# padding line 36
# padding line 37
# padding line 38
# padding line 39
# padding line 40
# padding line 41
# padding line 42
# padding line 43
# padding line 44
# padding line 45
# padding line 46
# padding line 47
# padding line 48
# padding line 49
# padding line 50
# padding line 51
# padding line 52
# padding line 53
# padding line 54
# padding line 55
# padding line 56
# padding line 57
# padding line 58
# padding line 59
# padding line 60
# padding line 61
# padding line 62
# padding line 63
# padding line 64
# padding line 65
# padding line 66
# padding line 67
# padding line 68