散戶反向策略：融資、當沖、集保數據的賺錢密碼

import numpy as np
from finlab import data
 
adj_close = data.get("etl:adj_close")
foreign_buy = data.get("institutional_investors_trading_summary:外陸資買賣超股數(不含外資自營商)")
trust_buy = data.get("institutional_investors_trading_summary:投信買賣超股數")
 
# 對齊 index 和 columns
common_idx = adj_close.index.intersection(foreign_buy.index).intersection(trust_buy.index)
common_cols = adj_close.columns.intersection(foreign_buy.columns).intersection(trust_buy.columns)
adj = adj_close.loc[common_idx, common_cols]
foreign_a = foreign_buy.loc[common_idx, common_cols]
trust_a = trust_buy.loc[common_idx, common_cols]
 
# 法人 60 日累計買賣超
institutional_net = foreign_a.rolling(60).sum() + trust_a.rolling(60).sum()
 
# 前瞻報酬
fwd_return = adj.shift(-21) / adj.shift(-1) - 1
 
# 市場基準：等權平均
market_return = fwd_return.mean(axis=1)
 
# 超額報酬 = 個股 - 市場
excess_return = fwd_return.sub(market_return, axis=0)
 
# 每年每月取樣，法人買超前 20% vs 後 20%
inst_yearly, retail_yearly = [], []
for year in range(2015, 2026):
    inst_year = institutional_net.loc[f'{year}-01-01':f'{year}-12-31']
    exc_year = excess_return.loc[f'{year}-01-01':f'{year}-12-31']
 
    inst_excess, retail_excess = [], []
    for i in range(0, len(inst_year), 20):  # 每月取樣一次
        if i >= len(exc_year):
            break
        row = inst_year.iloc[i].dropna()
        exc_row = exc_year.iloc[i].dropna()
        common = row.index.intersection(exc_row.index)
        if len(common) < 50:
            continue
        row, exc_row = row[common], exc_row[common]
        top_q = row.quantile(0.8)
        bot_q = row.quantile(0.2)
        if top_q == bot_q:
            continue
        inst_excess.append(exc_row[row >= top_q].mean())    # 法人組
        retail_excess.append(exc_row[row <= bot_q].mean())   # 散戶組
 
    inst_yearly.append(np.nanmean(inst_excess) * 12 * 100 if inst_excess else np.nan)
    retail_yearly.append(np.nanmean(retail_excess) * 12 * 100 if retail_excess else np.nan)

import pandas as pd
import numpy as np
from finlab import data
 
adj_close = data.get("etl:adj_close")
volume = data.get("price:成交股數")
daytrade_vol = data.get("intraday_trading:當日沖銷交易成交股數")
 
# 當沖比例 = 當沖成交股數 / 總成交股數（20 日均值平滑）
dt_ratio = daytrade_vol.rolling(20).mean() / volume.rolling(20).mean()
 
# 前瞻報酬
fwd_return = adj_close.shift(-21) / adj_close.shift(-1) - 1
 
# 市場基準 & 超額報酬
market_return = fwd_return.mean(axis=1)
excess_return = fwd_return.sub(market_return, axis=0)
 
# 每月取樣，按當沖比例分 10 組
group_returns = {i: [] for i in range(10)}
for i in range(0, len(dt_ratio), 20):
    row_dt = dt_ratio.iloc[i].dropna()
    row_exc = excess_return.iloc[i].dropna()
    common = row_dt.index.intersection(row_exc.index)
    row_dt, row_exc = row_dt[common], row_exc[common]
    row_dt = row_dt[row_dt > 0]       # 只看有當沖的股票
    row_exc = row_exc[row_dt.index]
    if len(row_dt) < 100:
        continue
    labels_q = pd.qcut(row_dt, 10, labels=False, duplicates='drop')
    for g in range(labels_q.max() + 1):
        group_returns[g].append(row_exc[labels_q == g].mean())
 
# 年化超額報酬
for g in range(10):
    annual = np.nanmean(group_returns[g]) * 12 * 100
    print(f"第{g+1}組: {annual:+.2f}%")

import pandas as pd
import numpy as np
from finlab import data
 
adj_close = data.get("etl:adj_close")
small_holder = data.get("etl:inventory:小於十張佔比")
 
# 散戶持股 4 週變化
small_change = small_holder - small_holder.shift(4)
 
# 前瞻報酬 & 超額報酬
fwd_return = adj_close.shift(-21) / adj_close.shift(-1) - 1
market_return = fwd_return.mean(axis=1)
excess_return = fwd_return.sub(market_return, axis=0)
 
# 每月取樣，按散戶持股變化分 10 組
group_returns = {i: [] for i in range(10)}
for i in range(0, len(small_change), 20):
    row_sc = small_change.iloc[i].dropna()
    row_exc = excess_return.iloc[i].dropna()
    common = row_sc.index.intersection(row_exc.index)
    row_sc, row_exc = row_sc[common], row_exc[common]
    if len(row_sc) < 100:
        continue
    labels_q = pd.qcut(row_sc, 10, labels=False, duplicates='drop')
    for g in range(labels_q.max() + 1):
        group_returns[g].append(row_exc[labels_q == g].mean())
 
# 年化超額報酬
for g in range(10):
    annual = np.nanmean(group_returns[g]) * 12 * 100
    print(f"第{g+1}組: {annual:+.2f}%")

from finlab import data
 
# 注意：0050 有配息，必須用還原股價
adj_close = data.get("etl:adj_close")
margin_balance = data.get("margin_transactions:融資今日餘額")
 
taiex = adj_close['0050'].dropna()
margin_total = margin_balance.sum(axis=1).dropna()
 
# 雙軸折線圖
fig, ax1 = plt.subplots(figsize=(12, 6))
ax1.plot(taiex.index, taiex.values, color='#2563EB', label='0050 還原股價')
ax2 = ax1.twinx()
ax2.fill_between(margin_total.index, margin_total.values / 1e12,
                 alpha=0.3, color='#EF4444')
# ... 標註事件、設定標題