""" 0050 annualised return, monthly investing (定期定額) and the best day of the month to invest (FinLab research, 2026-11). Data: Taiwan funds use etl:adj_close (dividends reinvested, 0050's 1-for-4 split on 2025-06-18 already adjusted). The price-only version of 0050 uses price:收盤價 with prices before 2025-06-18 divided by 4. QQQ and VT use us_fund_price:adj_close (USD, no FX). Every fund only uses data from its own first price in FinLab data. 1. Annualised return calendar years year-end close / previous year-end close - 1 (2007 from 04-23, 2026 to END) trailing N years from the last trading day on or before END minus N years holding N years buy at every trading day's close, sell at the first close on or after the same date N years later; no fees or tax 2. 0050 monthly investing: NT$10,000 on the 6th of every month (next trading day if closed), bought at the close; fee 0.1425% rounded down, at least NT$1, taken out of the 10,000; fractional shares; dividends reinvested through the adjusted price. Start in any month from 2007-05, invest for 5 / 10 / 15 years, value the holding at the close of the (12N + 1)-th debit day without selling. IRR = XIRR of the dated cash flows. 3. Day of the month: the same plan on day 1 ... 28 for 0050, 006208, 00878 and 00646, debit months from the first full month after listing to LAST_DEBIT_MONTH, valued at END. "Day d beats day 1" only if all hold: 1. plateau: d and its neighbours (d-1, d+1; 2-4 for d = 2, 26-28 for d = 28) x (whole period, first half, second half) = 9 cases, at least 7 with a higher IRR, 2. halves: debit months split in two, each half invested from scratch (first half valued at the last trading day of the following month, second half at END), d higher in both, 3. fees: with a NT$20 minimum fee instead of NT$1, d still higher over the whole period. "Day 1 beats day d" is the same three checks the other way round. A day is called better in general only if it passes in at least 3 of the 4 funds. Run: python research.py writes metrics.json and data.csv next to this file """ import json import math import operator from functools import cache from pathlib import Path import numpy as np import pandas as pd from finlab import data from scipy.optimize import brentq END = "2026-11-13" OUT = Path(__file__).resolve().parent TW_FUNDS = ("0050", "006208", "0056", "00878", "00692", "00646", "00662", "2330") US_FUNDS = ("QQQ", "VT") SPLIT_0050 = "2025-06-18" # 1-for-4 split, trading resumed this day RECENT_DAYS = ("2026-10-06", "2026-11-06", END) # closes quoted in the article TRAILING_YEARS = (5, 10, 15) HOLD_YEARS = (1, 3, 5, 10, 15) DCA_YEARS = (5, 10, 15) DCA_DAY = 6 DCA_FIRST_MONTH = "2007-05" # 0050 prices start 2007-04-23 AMOUNT = 10_000 FEE_RATE = 0.001425 MIN_FEE = 1 STRESS_MIN_FEE = 20 DAY_FUNDS = ("0050", "006208", "00878", "00646") DAYS = range(1, 29) PLATEAU_MIN_CASES = 7 # of the 9 cases in check 1 MIN_FUNDS = 3 # of the 4 DAY_FUNDS BROKER_DAYS = (6, 16, 26) LAST_DEBIT_MONTH = "2026-10" ROLLING_YEARS = 5 data.truncate_end = END # --------------------------------------------------------------------------- # Prices # --------------------------------------------------------------------------- @cache def total_returns(): """Dividend-reinvested prices of TW_FUNDS then US_FUNDS, keyed by ticker.""" prices = {} sources = (("tw", "etl:adj_close", TW_FUNDS), ("us", "us_fund_price:adj_close", US_FUNDS)) for market, dataset, funds in sources: data.set_market(market) table = data.get(dataset) prices.update({fund: table[fund].dropna() for fund in funds}) return prices def price_only_0050(): data.set_market("tw") close = data.get("price:收盤價")["0050"].dropna().copy() close[close.index < SPLIT_0050] /= 4 return close # --------------------------------------------------------------------------- # 1. Annualised return # --------------------------------------------------------------------------- def annualise(start_value, end_value, days): return (end_value / start_value) ** (365.25 / days) - 1 def calendar_years(price): year_end = price.groupby(price.index.year).last() previous = year_end.shift(1, fill_value=price.iloc[0]) return (year_end / previous - 1).round(6) def period_return(price, start_day): end_day = price.index[-1] days = (end_day - start_day).days return { "start": str(start_day.date()), "end": str(end_day.date()), "years": round(days / 365.25, 2), "total": round(price[end_day] / price[start_day] - 1, 6), "cagr": round(annualise(price[start_day], price[end_day], days), 6), } def trailing(price, years): target = price.index[-1] - pd.DateOffset(years=years) if target < price.index[0]: return None return period_return(price, price.loc[:target].index[-1]) def holding(price, years): """Annualised return of buying at every close and selling N years later.""" days = price.index sell_pos = days.searchsorted(days + pd.DateOffset(years=years)) valid = sell_pos < len(days) if not valid.any(): return None buy, sell = days[valid], days[sell_pos[valid]] held = (sell - buy).days.to_numpy() ann = pd.Series(annualise(price[buy].to_numpy(), price[sell].to_numpy(), held), index=buy) losses = ann[ann < 0] return { "n": int(len(ann)), "first_buy": str(buy[0].date()), "last_buy": str(buy[-1].date()), "worst": round(ann.min(), 6), "worst_buy": str(ann.idxmin().date()), "median": round(ann.median(), 6), "best": round(ann.max(), 6), "best_buy": str(ann.idxmax().date()), "loss_share": round((ann < 0).mean(), 6), "loss_n": int(len(losses)), "loss_buys_by_year": {str(y): int(n) for y, n in losses.groupby(losses.index.year).size().items()}, "loss_first_buy": str(losses.index[0].date()) if len(losses) else None, "loss_last_buy": str(losses.index[-1].date()) if len(losses) else None, } def max_drawdown(price): drawdown = price / price.cummax() - 1 trough = drawdown.idxmin() return {"mdd": round(drawdown.min(), 6), "trough": str(trough.date()), "peak": str(price.loc[:trough].idxmax().date())} def annual_return_section(): total, price_only = total_returns()["0050"], price_only_0050() section = {} for name, price in (("total", total), ("price_only", price_only)): section[name] = { "calendar_years": {str(y): r for y, r in calendar_years(price).items()}, "trailing": {str(n): trailing(price, n) for n in TRAILING_YEARS}, "full": period_return(price, price.index[0]), "holding": {str(n): holding(price, n) for n in HOLD_YEARS}, "max_drawdown": max_drawdown(price), } peers = {} for fund, price in total_returns().items(): if fund == "0050": continue start, end = price.index[0], price.index[-1] same_window = total.loc[start:end] peers[fund] = { "trailing": {str(n): trailing(price, n) for n in (5, 10)}, "full": period_return(price, start), "max_drawdown": max_drawdown(price), "hold_5y": holding(price, 5), "0050_same_window": period_return(same_window, same_window.index[0]), "0050_same_window_mdd": max_drawdown(same_window), } section["peers"] = peers section["recent_close"] = {day: round(price_only[day], 2) for day in RECENT_DAYS} section["trailing_5y_by_end"] = {day: trailing(total.loc[:day], 5)["cagr"] for day in RECENT_DAYS} return section # --------------------------------------------------------------------------- # 2 and 3. Monthly investing # --------------------------------------------------------------------------- def xirr(dates, flows): years = np.array([(d - dates[0]).days / 365 for d in dates]) flows = np.asarray(flows, dtype=float) return brentq(lambda r: np.sum(flows / (1 + r) ** years), -0.99, 10) def debit_day(days, month, day): """Trading day on or after `day` of `month` (pd.Period).""" target = pd.Timestamp(year=month.year, month=month.month, day=day) pos = days.searchsorted(target) return days[pos] if pos < len(days) else None def invest(price, debit_days, value_day, min_fee=MIN_FEE): fee = max(min_fee, math.floor(AMOUNT * FEE_RATE)) shares = sum((AMOUNT - fee) / price[d] for d in debit_days) value = shares * price[value_day] dates = [*debit_days, value_day] flows = [-AMOUNT] * len(debit_days) + [value] return {"invested": AMOUNT * len(debit_days), "value": value, "irr": xirr(dates, flows)} def last_day_of(days, month): in_month = days[(days >= month.start_time) & (days <= month.end_time)] return in_month[-1] if len(in_month) else None def row(section, fund, param, period, item, value): """One line of data.csv.""" return {"section": section, "fund": fund, "param": param, "period": period, "item": item, "value": value} def dca_0050(): price = total_returns()["0050"] days = price.index months = pd.period_range(DCA_FIRST_MONTH, pd.Period(END, "M"), freq="M") # debit day of each month; months whose debit day is after the last price drop off the end schedule = [d for d in (debit_day(days, m, DCA_DAY) for m in months) if d is not None] rows, summary = [], {} for years in DCA_YEARS: n = 12 * years results = [] for i in range(len(schedule) - n): res = invest(price, schedule[i:i + n], schedule[i + n]) results.append({"start": str(months[i]), "end": str(schedule[i + n].date()), **res}) frame = pd.DataFrame(results) rows += [row("dca_0050", "0050", f"{years}y", r["start"], "value", round(r["value"])) for r in results] rows += [row("dca_0050", "0050", f"{years}y", r["start"], "irr", round(r["irr"], 6)) for r in results] worst, best = frame.loc[frame["value"].idxmin()], frame.loc[frame["value"].idxmax()] summary[str(years)] = { "n": len(frame), "first_start": frame["start"].iloc[0], "last_start": frame["start"].iloc[-1], "invested": AMOUNT * n, "value_worst": round(worst["value"]), "value_worst_start": worst["start"], "value_median": round(frame["value"].median()), "value_best": round(best["value"]), "value_best_start": best["start"], "irr_worst": round(frame["irr"].min(), 6), "irr_median": round(frame["irr"].median(), 6), "irr_best": round(frame["irr"].max(), 6), "loss_share": round((frame["value"] < AMOUNT * n).mean(), 6), } to_date = invest(price, schedule, days[-1]) summary["since_2007_05"] = { "debits": len(schedule), "last_debit": str(schedule[-1].date()), "invested": to_date["invested"], "value": round(to_date["value"]), "irr": round(to_date["irr"], 6), } return summary, rows def three_checks(irr, stress, d, beats): """Checks 1-3 of the docstring for day d against day 1. beats = operator.gt asks "day d beats day 1"; operator.lt asks "day 1 beats day d". Returns the number of plateau cases that pass and the three results. """ neighbours = [2, 3, 4] if d == 2 else [26, 27, 28] if d == 28 else [d - 1, d, d + 1] cases = sum(beats(irr[p][nd], irr[p][1]) for nd in neighbours for p in irr) halves = beats(irr["first"][d], irr["first"][1]) and beats(irr["second"][d], irr["second"][1]) return cases, [cases >= PLATEAU_MIN_CASES, halves, beats(stress[d], stress[1])] def debit_day_fund(fund): price = total_returns()[fund] days = price.index first_month = pd.Period(days[0], "M") + 1 months = list(pd.period_range(first_month, LAST_DEBIT_MONTH, freq="M")) half = len(months) // 2 first, second = months[:half], months[half:] schedule = {d: [debit_day(days, m, d) for m in months] for d in DAYS} periods = { "full": (slice(None), days[-1]), "first": (slice(None, half), last_day_of(days, first[-1] + 1)), "second": (slice(half, None), days[-1]), } res = {p: {d: invest(price, schedule[d][part], vd) for d in DAYS} for p, (part, vd) in periods.items()} stress = {d: invest(price, schedule[d], days[-1], STRESS_MIN_FEE)["irr"] for d in DAYS} n_roll = 12 * ROLLING_YEARS rolling = {d: [] for d in DAYS} for i in range(len(months) - n_roll + 1): value_day = last_day_of(days, months[i + n_roll - 1] + 1) for d in DAYS: rolling[d].append(invest(price, schedule[d][i:i + n_roll], value_day)["irr"]) rolling = {d: np.array(v) for d, v in rolling.items()} irr = {p: {d: res[p][d]["irr"] for d in DAYS} for p in periods} value = {d: res["full"][d]["value"] for d in DAYS} per_day, rows = {}, [] for d in DAYS: if d == 1: verdict, checks = "base", None else: higher, better = three_checks(irr, stress, d, operator.gt) lower, worse = three_checks(irr, stress, d, operator.lt) verdict = "day" if all(better) else "day1" if all(worse) else "none" checks = {"better": better, "worse": worse, "higher_cases": higher, "lower_cases": lower} per_day[str(d)] = { "irr": round(irr["full"][d], 6), "irr_gap_vs_day1": round(irr["full"][d] - irr["full"][1], 6), "value": round(value[d]), "value_gap_vs_day1": round(value[d] - value[1]), "irr_first": round(irr["first"][d], 6), "irr_second": round(irr["second"][d], 6), "irr_min_fee_20": round(stress[d], 6), "rolling_win_vs_day1": None if d == 1 else round(float((rolling[d] > rolling[1]).mean()), 6), "rolling_mean_gap_vs_day1": round(float((rolling[d] - rolling[1]).mean()), 6), "verdict": verdict, "checks": checks, } for p in periods: rows.append(row("debit_day", fund, f"day_{d}", p, "irr", round(irr[p][d], 6))) rows.append(row("debit_day", fund, f"day_{d}", "full", "value", round(value[d]))) full_irr = pd.Series(irr["full"]) summary = { "months": [str(months[0]), str(months[-1])], "debits": len(months), "invested": AMOUNT * len(months), "first_half": [str(first[0]), str(first[-1]), str(periods["first"][1].date())], "second_half": [str(second[0]), str(second[-1]), str(periods["second"][1].date())], "rolling_windows": len(rolling[1]), "best_day": int(full_irr.idxmax()), "worst_day": int(full_irr.idxmin()), "irr_spread": round(full_irr.max() - full_irr.min(), 6), "value_spread": round(max(value.values()) - min(value.values())), "days_better_than_day1": [d for d in DAYS if per_day[str(d)]["verdict"] == "day"], "days_worse_than_day1": [d for d in DAYS if per_day[str(d)]["verdict"] == "day1"], "days": per_day, } return summary, rows def debit_day_section(): funds, rows = {}, [] for fund in DAY_FUNDS: funds[fund], fund_rows = debit_day_fund(fund) rows += fund_rows passes = {d: sum(d in funds[f]["days_better_than_day1"] for f in DAY_FUNDS) for d in DAYS} general = [d for d, n in passes.items() if n >= MIN_FUNDS] return {"funds": funds, "passes_per_day": passes, "better_in_3_of_4": general, "broker_days": list(BROKER_DAYS)}, rows def main(): annual = annual_return_section() rows = [] for name in ("total", "price_only"): rows += [row(f"calendar_year_{name}", "0050", "", y, "return", r) for y, r in annual[name]["calendar_years"].items()] dca, dca_rows = dca_0050() days, day_rows = debit_day_section() metrics = { "data_end": END, "last_price": {f: str(price.index[-1].date()) for f, price in total_returns().items()}, "annual_return": annual, "dca_0050": dca, "debit_day": days, } text = json.dumps(metrics, ensure_ascii=False, indent=1, default=lambda value: value.item()) (OUT / "metrics.json").write_text(text + "\n") pd.DataFrame(rows + dca_rows + day_rows).to_csv(OUT / "data.csv", index=False) print(json.dumps({"better_in_3_of_4": days["better_in_3_of_4"], "passes_per_day": days["passes_per_day"]})) if __name__ == "__main__": main()