2024年05月20日玄貓（BlackCat）

Pandas 進階資料處理技術：管道操作、Top N 分析與金融應用實務

深入探討 Pandas 的進階資料處理技術，涵蓋管道操作模式、Top N 分析方法、金融市場追蹤停損策略實作，以及運動數據分析應用，提供台灣資料科學與金融科技從業人員完整的實務操作指南

資料科學 Python 程式設計金融科技

Pandas Python 管道操作 pipe nlargest nsmallest Top N 分析金融資料分析追蹤停損資料篩選 idxmax idxmin cummax cummin 方法鏈結 Data Pipeline Financial Analysis

在現代資料科學的工作流程中，程式碼的可讀性與維護性往往與運算效能同等重要。Pandas 作為 Python 資料分析生態系統的核心函式庫，不僅提供了強大的資料處理能力，更透過精心設計的方法鏈結（Method Chaining）與管道操作（Pipeline）模式，讓開發者能夠撰寫優雅且易於理解的資料轉換流程。對於台灣的資料科學從業人員而言，無論是處理金融市場資料、電商交易記錄，或是社群媒體分析，掌握這些進階技巧都能顯著提升工作效率。

管道操作的核心理念在於將複雜的資料處理流程分解為一系列簡單、可重用的轉換函式，透過 pipe 方法串接這些函式，形成清晰的資料處理管道。這種設計模式源自於 Unix 系統的管道哲學，每個處理步驟專注於單一職責，透過標準化的輸入輸出介面組合成複雜的工作流程。在實務應用中，這種模式不僅讓程式碼更容易測試與除錯，更重要的是讓資料轉換的邏輯一目了然，新進團隊成員能夠快速理解程式碼的運作方式。

Top N 分析是資料探索階段的常用技巧，從海量資料中快速找出最大或最小的前幾筆記錄，能夠幫助我們快速掌握資料的分布特性。Pandas 提供的 nlargest 與 nsmallest 方法不僅在語意上更為明確，在效能上也經過最佳化，相較於先排序再取前 N 筆的傳統做法更為高效。在台灣的商業應用中，這類分析隨處可見，例如電商平台找出銷售額最高的商品、人力資源部門分析薪資水準的前後段分布，或是投資分析師篩選表現最佳的投資標的。

金融市場的追蹤停損策略展現了 Pandas 在時間序列資料處理上的強大能力。累積最大值（cummax）與累積最小值（cummin）方法能夠動態追蹤歷史極值，這在風險管理與交易策略中扮演關鍵角色。台灣投資人在操作台股、美股或加密貨幣時，往往需要設定停損點以控制風險，而程式化的追蹤停損機制能夠自動化這個過程，避免人為情緒干擾投資決策。本文將從基礎的管道操作開始，逐步深入 Top N 分析技巧，最後透過實際的金融應用案例，展示這些技術如何在真實世界中創造價值。

Pandas 管道操作的設計哲學與實務應用

管道操作模式的核心價值在於提升程式碼的組織性與可讀性。在傳統的資料處理流程中，開發者經常需要建立大量的中間變數來儲存每個轉換步驟的結果，這不僅使程式碼冗長，更容易造成變數命名的困擾與記憶體的浪費。透過 pipe 方法，我們可以將資料轉換流程表達為一連串的函式呼叫，每個函式接收 DataFrame 作為輸入並回傳轉換後的 DataFrame，形成流暢的資料處理管道。

在實務應用中，管道操作特別適合處理需要多步驟轉換的資料清洗工作。以台灣的電商資料為例，原始訂單資料可能需要經過幣別轉換、缺失值處理、異常值過濾、欄位重新命名等多個步驟才能用於後續分析。傳統做法可能會建立 df_step1、df_step2、df_step3 等一系列中間變數，不僅占用記憶體空間，程式碼的閱讀也需要不斷跳躍。管道操作讓整個轉換流程變成線性的函式鏈結，從資料讀取到最終結果一氣呵成。

pipe 方法的另一個重要特性是支援攜帶額外參數的函式。在資料轉換過程中，某些處理邏輯需要根據外部條件調整行為，例如根據不同的匯率轉換幣別、依據不同的閾值過濾異常值，或是按照不同的格式標準化文字欄位。pipe 方法允許我們在呼叫時傳遞這些額外參數，保持了函式的通用性與彈性。這種設計讓同一套資料處理函式能夠適應不同的業務場景，大幅提升了程式碼的重用性。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Pandas 管道操作實務應用
展示如何使用 pipe 方法建立清晰的資料處理流程，
以及如何處理需要額外參數的轉換函式
"""

import pandas as pd
import numpy as np
from typing import Literal

# ============================================================================
# 範例一：基礎管道操作 - 建立可重用的資料轉換函式
# ============================================================================

def convert_currency_column(
    df: pd.DataFrame,
    column_name: str,
    exchange_rate: float
) -> pd.DataFrame:
    """
    轉換指定欄位的幣別
    
    Parameters:
    -----------
    df : pd.DataFrame
        輸入的 DataFrame
    column_name : str
        要轉換的欄位名稱
    exchange_rate : float
        匯率（例如 USD to TWD = 31.5）
    
    Returns:
    --------
    pd.DataFrame
        幣別轉換後的 DataFrame
    """
    # 使用 assign 方法建立新欄位，保持原始資料不變
    # 這種不可變（immutable）的設計模式讓程式碼更安全
    return df.assign(
        **{column_name: df[column_name] * exchange_rate}
    )

def filter_by_threshold(
    df: pd.DataFrame,
    column_name: str,
    threshold: float,
    keep: Literal['above', 'below'] = 'above'
) -> pd.DataFrame:
    """
    根據閾值篩選資料
    
    Parameters:
    -----------
    df : pd.DataFrame
        輸入的 DataFrame
    column_name : str
        要篩選的欄位名稱
    threshold : float
        閾值
    keep : Literal['above', 'below']
        保留高於或低於閾值的資料
    
    Returns:
    --------
    pd.DataFrame
        篩選後的 DataFrame
    """
    if keep == 'above':
        # 保留大於閾值的資料
        mask = df[column_name] > threshold
    else:
        # 保留小於閾值的資料
        mask = df[column_name] < threshold
    
    return df[mask].copy()

def add_percentage_column(
    df: pd.DataFrame,
    numerator_col: str,
    denominator_col: str,
    result_col: str
) -> pd.DataFrame:
    """
    計算百分比並新增為新欄位
    
    Parameters:
    -----------
    df : pd.DataFrame
        輸入的 DataFrame
    numerator_col : str
        分子欄位名稱
    denominator_col : str
        分母欄位名稱
    result_col : str
        結果欄位名稱
    
    Returns:
    --------
    pd.DataFrame
        新增百分比欄位後的 DataFrame
    """
    # 計算百分比，避免除以零的錯誤
    percentage = (df[numerator_col] / df[denominator_col] * 100).fillna(0)
    
    return df.assign(**{result_col: percentage})

# 建立範例資料：台灣電商訂單資料（美金計價）
orders_data = {
    '訂單編號': ['ORD001', 'ORD002', 'ORD003', 'ORD004', 'ORD005'],
    '商品名稱': ['筆記型電腦', '智慧型手機', '平板電腦', '耳機', '滑鼠'],
    '訂單金額_USD': [1200.00, 800.00, 450.00, 150.00, 25.00],
    '運費_USD': [50.00, 30.00, 20.00, 10.00, 5.00],
    '折扣金額_USD': [100.00, 50.00, 30.00, 10.00, 2.00]
}

orders_df = pd.DataFrame(orders_data)

print("原始訂單資料（美金計價）：")
print(orders_df)
print("\n" + "="*70 + "\n")

# ============================================================================
# 傳統做法：使用多個中間變數
# ============================================================================

# 步驟一：轉換為台幣（假設匯率 1 USD = 31.5 TWD）
orders_step1 = orders_df.copy()
orders_step1['訂單金額_TWD'] = orders_step1['訂單金額_USD'] * 31.5
orders_step1['運費_TWD'] = orders_step1['運費_USD'] * 31.5
orders_step1['折扣金額_TWD'] = orders_step1['折扣金額_USD'] * 31.5

# 步驟二：計算實付金額
orders_step2 = orders_step1.copy()
orders_step2['實付金額_TWD'] = (
    orders_step2['訂單金額_TWD'] + 
    orders_step2['運費_TWD'] - 
    orders_step2['折扣金額_TWD']
)

# 步驟三：篩選高於 10000 台幣的訂單
orders_step3 = orders_step2[orders_step2['實付金額_TWD'] > 10000].copy()

print("傳統做法結果（需要 3 個中間變數）：")
print(orders_step3[['訂單編號', '商品名稱', '實付金額_TWD']])
print("\n" + "="*70 + "\n")

# ============================================================================
# 管道操作做法：清晰的資料處理流程
# ============================================================================

# 定義資料處理函式
def convert_to_twd(df: pd.DataFrame, rate: float = 31.5) -> pd.DataFrame:
    """將美金欄位轉換為台幣"""
    return df.assign(
        訂單金額_TWD=df['訂單金額_USD'] * rate,
        運費_TWD=df['運費_USD'] * rate,
        折扣金額_TWD=df['折扣金額_USD'] * rate
    )

def calculate_final_amount(df: pd.DataFrame) -> pd.DataFrame:
    """計算實付金額"""
    return df.assign(
        實付金額_TWD=df['訂單金額_TWD'] + df['運費_TWD'] - df['折扣金額_TWD']
    )

def filter_high_value_orders(
    df: pd.DataFrame,
    threshold: float = 10000
) -> pd.DataFrame:
    """篩選高價值訂單"""
    return df[df['實付金額_TWD'] > threshold].copy()

def select_summary_columns(df: pd.DataFrame) -> pd.DataFrame:
    """選取摘要欄位"""
    return df[['訂單編號', '商品名稱', '實付金額_TWD']]

# 使用管道操作串接所有轉換步驟
# 資料流向一目了然，無需中間變數
result = (
    orders_df
    .pipe(convert_to_twd, rate=31.5)
    .pipe(calculate_final_amount)
    .pipe(filter_high_value_orders, threshold=10000)
    .pipe(select_summary_columns)
)

print("管道操作做法結果（無中間變數，流程清晰）：")
print(result)
print("\n" + "="*70 + "\n")

# ============================================================================
# 範例二：處理台灣股市資料的管道操作
# ============================================================================

# 模擬台灣上市公司股價資料
stock_data = {
    '股票代碼': ['2330', '2317', '2454', '2412', '2882'],
    '公司名稱': ['台積電', '鴻海', '聯發科', '中華電', '國泰金'],
    '收盤價': [585.0, 105.5, 1125.0, 123.5, 45.8],
    '成交量': [45230, 38950, 12340, 8760, 15680],
    '本益比': [28.5, 15.2, 32.8, 18.9, 12.3]
}

stocks_df = pd.DataFrame(stock_data)

print("台灣股市資料：")
print(stocks_df)
print("\n" + "="*70 + "\n")

def calculate_market_cap_estimate(df: pd.DataFrame) -> pd.DataFrame:
    """
    估算市值（簡化計算，實際應使用流通股數）
    假設每家公司發行 1 億股
    """
    shares_outstanding = 100_000_000  # 1 億股
    return df.assign(
        市值估算_億=df['收盤價'] * shares_outstanding / 100_000_000
    )

def add_value_rating(df: pd.DataFrame) -> pd.DataFrame:
    """
    根據本益比評估價值
    本益比 < 15: 便宜
    15 <= 本益比 < 25: 合理
    本益比 >= 25: 昂貴
    """
    def rate_value(pe_ratio):
        if pe_ratio < 15:
            return '便宜'
        elif pe_ratio < 25:
            return '合理'
        else:
            return '昂貴'
    
    return df.assign(
        價值評級=df['本益比'].apply(rate_value)
    )

def sort_by_metric(
    df: pd.DataFrame,
    column: str,
    ascending: bool = False
) -> pd.DataFrame:
    """依指定欄位排序"""
    return df.sort_values(column, ascending=ascending).reset_index(drop=True)

# 建立股票分析管道
stock_analysis = (
    stocks_df
    .pipe(calculate_market_cap_estimate)
    .pipe(add_value_rating)
    .pipe(sort_by_metric, column='市值估算_億', ascending=False)
)

print("股票分析結果（依市值排序）：")
print(stock_analysis)
print("\n" + "="*70 + "\n")

# ============================================================================
# 範例三：條件式管道操作
# ============================================================================

def apply_discount_if_applicable(
    df: pd.DataFrame,
    apply_vip_discount: bool = False
) -> pd.DataFrame:
    """
    根據條件決定是否套用 VIP 折扣
    展示管道中的條件邏輯處理
    """
    if apply_vip_discount:
        # VIP 客戶享有 95 折
        return df.assign(
            實付金額_TWD=df['實付金額_TWD'] * 0.95,
            折扣類型='VIP 折扣'
        )
    else:
        # 一般客戶無額外折扣
        return df.assign(折扣類型='無')

# 示範條件式管道：VIP 客戶與一般客戶的不同處理流程
is_vip_customer = True

vip_result = (
    orders_df
    .pipe(convert_to_twd, rate=31.5)
    .pipe(calculate_final_amount)
    .pipe(apply_discount_if_applicable, apply_vip_discount=is_vip_customer)
    .pipe(select_summary_columns)
)

print(f"VIP 客戶訂單處理結果（VIP={is_vip_customer}）：")
print(vip_result)

管道操作的進階應用包括條件式處理與動態函式組合。在某些場景下，資料處理流程需要根據外部條件決定是否執行某些轉換步驟，或是選擇不同的處理邏輯。透過 Python 的條件式語法與函式式程式設計技巧，我們可以動態組合管道中的函式序列。例如在處理不同層級的客戶訂單時，VIP 客戶可能享有額外折扣，管道操作可以優雅地整合這種條件邏輯，保持程式碼的可讀性。

錯誤處理是管道操作中不可忽視的面向。當管道中的某個函式發生異常時，整個處理流程會中斷，因此每個轉換函式都應該包含適當的錯誤處理邏輯。在台灣的金融資料處理場景中，資料品質問題時有所聞，例如股價資料可能包含異常值、交易量可能為零、某些欄位可能缺失。穩健的管道函式應該能夠檢測這些問題並採取適當的處理策略，例如記錄警告訊息、使用預設值填充，或是將問題資料隔離到另一個 DataFrame 中供後續人工檢視。

@startuml
!define PLANTUML_FORMAT svg
!theme _none_

skinparam dpi auto
skinparam shadowing false
skinparam linetype ortho
skinparam roundcorner 5
skinparam defaultFontName "Microsoft JhengHei UI"
skinparam defaultFontSize 14
skinparam minClassWidth 150

participant "原始資料\nDataFrame" as Raw
participant "pipe 方法" as Pipe
participant "轉換函式 1\n幣別轉換" as Func1
participant "轉換函式 2\n金額計算" as Func2
participant "轉換函式 3\n資料篩選" as Func3
participant "轉換函式 4\n欄位選取" as Func4
participant "最終結果\nDataFrame" as Result

Raw -> Pipe: df.pipe(convert_to_twd)
activate Pipe
Pipe -> Func1: 傳入 DataFrame\n+ 額外參數 (rate=31.5)
activate Func1
Func1 -> Func1: 執行幣別轉換\nUSD -> TWD
Func1 -> Func1: 使用 assign 建立新欄位\n保持不可變性
Func1 --> Pipe: 回傳轉換後的 DataFrame
deactivate Func1

Pipe -> Func2: 傳入上一步的結果
activate Func2
Func2 -> Func2: 計算實付金額\n訂單金額 + 運費 - 折扣
Func2 -> Func2: 使用 assign 新增欄位
Func2 --> Pipe: 回傳計算後的 DataFrame
deactivate Func2

Pipe -> Func3: 傳入上一步的結果\n+ 閾值參數
activate Func3
Func3 -> Func3: 建立布林遮罩\n實付金額 > 10000
Func3 -> Func3: 篩選符合條件的資料
Func3 --> Pipe: 回傳篩選後的 DataFrame
deactivate Func3

Pipe -> Func4: 傳入上一步的結果
activate Func4
Func4 -> Func4: 選取摘要欄位\n訂單編號、商品、金額
Func4 --> Pipe: 回傳精簡的 DataFrame
deactivate Func4

Pipe --> Result: 管道處理完成\n回傳最終結果
deactivate Pipe

note over Raw, Result
  管道操作的優勢：
  1. 無需中間變數，節省記憶體
  2. 資料流向清晰，易於理解
  3. 函式可重用，提升程式碼品質
  4. 支援參數傳遞，保持彈性
  5. 符合函數式程式設計原則
end note

note right of Func1
  每個轉換函式遵循
  相同的介面規範：
  - 輸入：DataFrame
  - 輸出：DataFrame
  - 不修改原始資料
end note

@enduml

Top N 分析的高效實作策略

Top N 分析是資料探索階段最常用的技術之一，從大量資料中快速識別極值樣本能夠幫助我們理解資料分布、發現異常模式，或是聚焦於最重要的觀察對象。Pandas 提供的 nlargest 與 nsmallest 方法不僅在語意上更為清晰，在演算法實作上也經過精心最佳化。相較於傳統的全域排序後取前 N 筆做法，這兩個方法使用部分排序演算法，時間複雜度從 O(n log n) 降低至 O(n log k)，其中 k 是要取的元素數量。當資料量龐大而 k 相對較小時，效能提升極為顯著。

在台灣的商業分析場景中，Top N 分析的應用極為廣泛。電商平台需要找出銷售額最高的商品以規劃促銷策略，人力資源部門分析員工薪資分布時會關注最高薪與最低薪的群體，投資分析師篩選股票時往往先從市值最大或成長最快的標的開始研究。這些場景的共通點在於我們不需要完整的排序結果，只需要前幾名或後幾名的資料，nlargest 與 nsmallest 正好滿足這個需求。

複合條件的 Top N 分析展現了這類方法的靈活性。我們可以先使用某個條件篩選資料子集，再在子集中尋找極值。例如在分析台灣電影資料時，可能先篩選出評分超過 8 分的高評價電影，再從中找出票房最高的前十部。這種層次化的分析策略能夠發現更有價值的洞察，避免被整體資料的雜訊所干擾。管道操作與 Top N 分析的結合更是威力強大，可以建立複雜的資料探索流程。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Pandas Top N 分析實務應用
展示 nlargest 與 nsmallest 方法在不同場景的使用技巧，
以及如何結合管道操作進行複合條件分析
"""

import pandas as pd
import numpy as np

# ============================================================================
# 範例一：電影資料的 Top N 分析
# ============================================================================

# 建立台灣電影資料範例（虛構資料）
movies_data = {
    '電影名稱': [
        '海角七號', '那些年，我們一起追的女孩', '雞排英雄', 
        '陣頭', '艋舺', '賽德克·巴萊', '大尾鱸鰻',
        '露西', '女朋友·男朋友', '台北物語',
        '一頁台北', '總舖師', '翻滾吧！阿信', 
        '消失的情人節', '刻在你心底的名字'
    ],
    '上映年份': [
        2008, 2011, 2011, 2012, 2010, 2011, 2013,
        2014, 2012, 2018, 2010, 2013, 2011,
        2020, 2020
    ],
    'IMDb 評分': [
        7.5, 7.6, 7.2, 7.3, 6.8, 7.9, 5.8,
        6.4, 6.9, 6.5, 6.7, 7.1, 7.4,
        7.2, 7.3
    ],
    '預算_百萬': [
        50, 30, 25, 45, 80, 250, 40,
        150, 35, 60, 28, 55, 48,
        70, 65
    ],
    '票房_百萬': [
        530, 425, 235, 340, 265, 900, 395,
        580, 125, 180, 95, 320, 285,
        245, 320
    ]
}

movies_df = pd.DataFrame(movies_data)

print("台灣電影資料：")
print(movies_df)
print("\n" + "="*70 + "\n")

# ============================================================================
# 基礎 Top N 分析：找出評分最高的電影
# ============================================================================

# 找出 IMDb 評分最高的前 5 部電影
top_rated_movies = movies_df.nlargest(5, 'IMDb 評分')

print("評分最高的前 5 部電影：")
print(top_rated_movies[['電影名稱', 'IMDb 評分', '票房_百萬']])
print("\n" + "="*70 + "\n")

# 找出預算最低的前 3 部電影
lowest_budget_movies = movies_df.nsmallest(3, '預算_百萬')

print("預算最低的前 3 部電影：")
print(lowest_budget_movies[['電影名稱', '預算_百萬', '票房_百萬']])
print("\n" + "="*70 + "\n")

# ============================================================================
# 複合條件 Top N 分析：從高評分電影中找出票房冠軍
# ============================================================================

# 先篩選評分超過 7.0 的電影
high_rated = movies_df[movies_df['IMDb 評分'] >= 7.0]

# 在高評分電影中找出票房最高的前 3 部
top_grossing_among_high_rated = high_rated.nlargest(3, '票房_百萬')

print("高評分電影（≥7.0）中票房最高的前 3 部：")
print(top_grossing_among_high_rated[['電影名稱', 'IMDb 評分', '票房_百萬']])
print("\n" + "="*70 + "\n")

# ============================================================================
# 使用管道操作的 Top N 分析
# ============================================================================

def calculate_roi(df: pd.DataFrame) -> pd.DataFrame:
    """計算投資報酬率（ROI）"""
    return df.assign(
        ROI=((df['票房_百萬'] - df['預算_百萬']) / df['預算_百萬'] * 100).round(2)
    )

def filter_recent_movies(df: pd.DataFrame, year: int = 2010) -> pd.DataFrame:
    """篩選指定年份之後的電影"""
    return df[df['上映年份'] >= year].copy()

# 建立分析管道：找出 2010 年後 ROI 最高的電影
best_roi_movies = (
    movies_df
    .pipe(filter_recent_movies, year=2010)
    .pipe(calculate_roi)
    .nlargest(5, 'ROI')
    [['電影名稱', '上映年份', '預算_百萬', '票房_百萬', 'ROI']]
)

print("2010 年後投資報酬率最高的前 5 部電影：")
print(best_roi_movies)
print("\n" + "="*70 + "\n")

# ============================================================================
# 範例二：台灣上市公司市值分析
# ============================================================================

# 建立台灣上市公司資料（虛構資料，僅供示範）
companies_data = {
    '股票代碼': [
        '2330', '2317', '2454', '2412', '2882',
        '2881', '2886', '2308', '2303', '1301'
    ],
    '公司名稱': [
        '台積電', '鴻海', '聯發科', '中華電', '國泰金',
        '富邦金', '兆豐金', '台達電', '聯電', '台塑'
    ],
    '產業別': [
        '半導體', '電子製造', '半導體', '電信', '金融',
        '金融', '金融', '電子零組件', '半導體', '塑膠'
    ],
    '市值_億': [
        15000, 1850, 4200, 1560, 1320,
        1280, 980, 2100, 1950, 1480
    ],
    '本益比': [
        28.5, 15.2, 32.8, 18.9, 12.3,
        11.8, 10.5, 24.6, 19.8, 16.4
    ],
    '殖利率_%': [
        1.8, 4.2, 2.1, 5.5, 6.2,
        5.8, 6.5, 3.2, 4.8, 5.2
    ]
}

companies_df = pd.DataFrame(companies_data)

print("台灣主要上市公司資料：")
print(companies_df)
print("\n" + "="*70 + "\n")

# 找出市值最大的前 3 家公司
largest_companies = companies_df.nlargest(3, '市值_億')

print("市值最大的前 3 家公司：")
print(largest_companies[['公司名稱', '產業別', '市值_億']])
print("\n" + "="*70 + "\n")

# 找出殖利率最高的前 5 家公司（適合存股族）
highest_yield = companies_df.nlargest(5, '殖利率_%')

print("殖利率最高的前 5 家公司：")
print(highest_yield[['公司名稱', '殖利率_%', '本益比']])
print("\n" + "="*70 + "\n")

# 產業內分析：找出半導體產業中市值最大的公司
semiconductor_companies = companies_df[companies_df['產業別'] == '半導體']
top_semiconductor = semiconductor_companies.nlargest(3, '市值_億')

print("半導體產業市值前 3 大公司：")
print(top_semiconductor[['公司名稱', '市值_億', '本益比']])
print("\n" + "="*70 + "\n")

# ============================================================================
# 範例三：多欄位排序的 Top N 分析
# ============================================================================

# 找出本益比最低（價值被低估）且市值夠大（流動性佳）的標的
# 首先篩選市值超過 1000 億的公司
large_cap = companies_df[companies_df['市值_億'] > 1000]

# 在大型股中找出本益比最低的 3 家
value_stocks = large_cap.nsmallest(3, '本益比')

print("大型股（市值 > 1000 億）中本益比最低的 3 家：")
print(value_stocks[['公司名稱', '市值_億', '本益比', '殖利率_%']])
print("\n" + "="*70 + "\n")

# ============================================================================
# 範例四：同時取最大和最小值
# ============================================================================

# 分析極端值：市值最大與最小的公司對比
top_3_largest = companies_df.nlargest(3, '市值_億')
bottom_3_smallest = companies_df.nsmallest(3, '市值_億')

# 使用 concat 合併結果
market_cap_extremes = pd.concat([
    top_3_largest.assign(類別='市值前三大'),
    bottom_3_smallest.assign(類別='市值後三小')
])

print("市值極端值分析：")
print(market_cap_extremes[['類別', '公司名稱', '市值_億', '產業別']])

nlargest 與 nsmallest 方法支援多欄位排序，當第一個排序欄位出現相同值時，可以透過第二個欄位決定排序順序。這在實務上極為實用，例如在分析學生成績時，若總分相同則以國文成績高低決定名次。在台灣的房地產分析中，可能先以總價篩選，總價相同時再以坪數大小排序。這種多層次的排序邏輯讓資料分析更加精確，避免了因相同數值而產生的排名歧義。

效能考量是選擇 Top N 方法而非全域排序的關鍵因素。在處理包含數百萬筆記錄的資料集時，若僅需要前 100 筆資料，使用 nlargest 的效能優勢極為明顯。Pandas 內部使用堆積（Heap）資料結構實作部分排序，只維護大小為 k 的最小堆積或最大堆積，避免了不必要的比較操作。在台灣的大數據應用場景中，例如電信業者分析數百萬用戶的通話記錄、電商平台處理千萬筆交易資料，這種效能提升能夠直接轉化為更快的分析速度與更低的運算成本。

@startuml
!define PLANTUML_FORMAT svg
!theme _none_

skinparam dpi auto
skinparam shadowing false
skinparam linetype ortho
skinparam roundcorner 5
skinparam defaultFontName "Microsoft JhengHei UI"
skinparam defaultFontSize 14
skinparam minClassWidth 140

package "Top N 分析工作流程" {
  
  frame "資料準備階段" as Prep {
    rectangle "原始資料集" as Raw
    rectangle "資料清洗" as Clean
    rectangle "特徵工程" as Feature
  }
  
  frame "篩選階段" as Filter {
    rectangle "條件篩選" as CondFilter
    rectangle "產業分類" as Industry
    rectangle "時間範圍" as TimeRange
  }
  
  frame "Top N 分析" as TopN {
    card "nlargest\n取最大值" as NLargest
    card "nsmallest\n取最小值" as NSmallest
    card "多欄位排序" as MultiCol
  }
  
  frame "結果應用" as Result {
    usecase "投資決策" as Investment
    usecase "商品推薦" as Recommend
    usecase "風險管理" as Risk
    usecase "績效評估" as Performance
  }
  
  database "分析結果" as Output {
    [前 N 大標的]
    [後 N 小標的]
    [極端值比較]
  }
}

Raw --> Clean
Clean --> Feature
Feature --> CondFilter

CondFilter --> Industry
CondFilter --> TimeRange

Industry --> NLargest
Industry --> NSmallest
TimeRange --> NLargest
TimeRange --> NSmallest

NLargest --> MultiCol
NSmallest --> MultiCol

MultiCol --> Output

Output --> Investment
Output --> Recommend
Output --> Risk
Output --> Performance

note right of NLargest
  使用堆積演算法
  時間複雜度 O(n log k)
  k << n 時效能優異
end note

note right of NSmallest
  適用場景：
  - 找出異常值
  - 識別風險標的
  - 發掘被低估資產
end note

note bottom of Output
  Top N 結果可作為：
  1. 決策參考依據
  2. 進一步分析起點
  3. 視覺化呈現素材
end note

@enduml

金融市場追蹤停損策略的程式化實作

追蹤停損（Trailing Stop）是風險管理的重要工具，其核心概念是隨著資產價格上漲動態調整停損點，既保護獲利又給予價格上漲空間。在台灣的投資環境中，無論是交易台股、美股、期貨或加密貨幣，追蹤停損都是專業投資人常用的策略。Pandas 提供的累積極值方法（cummax 與 cummin）完美契合這個需求，能夠輕鬆實作追蹤停損的價格計算邏輯。

累積最大值（cummax）方法會回傳一個新的 Series，其中每個位置的值都是從序列開始到該位置的最大值。這個特性正好對應多頭部位的追蹤停損邏輯：停損價格應該追蹤歷史最高價的固定比例。舉例而言，若設定停損幅度為 10%，則停損價格永遠是歷史最高價的 90%。當股價創新高時，停損價格隨之提升。當股價回檔但未觸及停損價時，停損價格維持不變。這種單向調整的特性確保了獲利不會因小幅震盪而過早平倉。

空頭部位的追蹤停損邏輯則相反，需要追蹤歷史最低價。cummin 方法提供了累積最小值的計算能力，讓我們能夠實作空頭停損策略。在台灣期貨市場或融券交易中，投資人做空看跌的標的時，需要設定向上的停損點以控制損失。若設定停損幅度為 10%，則停損價格是歷史最低價的 110%。當價格持續下跌時停損價格跟著下降，但價格反彈時停損價格固定，避免了過早回補的遺憾。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
追蹤停損策略程式化實作
展示如何使用 Pandas 的 cummax、cummin 方法實作追蹤停損邏輯，
以及使用 idxmax、idxmin 找出停損觸發時點
"""

import pandas as pd
import numpy as np
from datetime import datetime, timedelta

# ============================================================================
# 範例一：台股追蹤停損策略
# ============================================================================

# 建立台積電（2330）股價模擬資料
# 假設從 2024/01/02 開始，共 60 個交易日
np.random.seed(42)
start_date = datetime(2024, 1, 2)
trading_days = pd.date_range(start=start_date, periods=60, freq='B')

# 模擬股價走勢（以 500 元為起點）
initial_price = 500.0
returns = np.random.normal(0.002, 0.02, size=60)  # 日報酬率
prices = initial_price * (1 + returns).cumprod()

# 建立價格 Series，使用日期作為索引
tsmc_prices = pd.Series(prices, index=trading_days, name='收盤價')

print("台積電股價走勢（模擬資料）：")
print(tsmc_prices.head(10))
print(f"...\n最後 5 天：")
print(tsmc_prices.tail(5))
print("\n" + "="*70 + "\n")

# ============================================================================
# 計算多頭追蹤停損價格
# ============================================================================

# 設定停損幅度為 10%（保護 90% 的獲利）
stop_loss_percentage = 0.10
protection_ratio = 1 - stop_loss_percentage

# 計算累積最高價
# cummax() 回傳每個時點之前（含）的最高價
cumulative_high = tsmc_prices.cummax()

# 計算追蹤停損價格
# 停損價 = 累積最高價 × 保護比例
trailing_stop_long = cumulative_high * protection_ratio

print("多頭追蹤停損價格計算：")
comparison_df = pd.DataFrame({
    '收盤價': tsmc_prices,
    '累積最高價': cumulative_high,
    '停損價（90%）': trailing_stop_long.round(2)
})
print(comparison_df.head(10))
print("\n" + "="*70 + "\n")

# ============================================================================
# 檢查停損觸發條件
# ============================================================================

# 多頭停損觸發條件：收盤價 <= 停損價
stop_triggered_long = tsmc_prices <= trailing_stop_long

print("停損觸發檢查（多頭部位）：")
print(f"總交易日數：{len(tsmc_prices)}")
print(f"觸發停損天數：{stop_triggered_long.sum()}")

# 找出第一次觸發停損的日期
# idxmax() 會回傳第一個 True 值的索引
if stop_triggered_long.any():
    first_trigger_date = stop_triggered_long.idxmax()
    trigger_price = tsmc_prices.loc[first_trigger_date]
    stop_price = trailing_stop_long.loc[first_trigger_date]
    
    print(f"\n第一次觸發停損日期：{first_trigger_date.strftime('%Y-%m-%d')}")
    print(f"當日收盤價：{trigger_price:.2f} 元")
    print(f"停損價格：{stop_price:.2f} 元")
    
    # 計算持有期間的報酬率
    initial_purchase_price = tsmc_prices.iloc[0]
    return_pct = ((trigger_price - initial_purchase_price) / 
                  initial_purchase_price * 100)
    print(f"持有報酬率：{return_pct:.2f}%")
else:
    print("\n在觀察期間內未觸發停損")

print("\n" + "="*70 + "\n")

# ============================================================================
# 範例二：空頭追蹤停損策略
# ============================================================================

# 假設投資人在某個時點開始做空台積電
# 空頭追蹤停損需要追蹤累積最低價

# 計算累積最低價
cumulative_low = tsmc_prices.cummin()

# 計算空頭停損價格
# 停損價 = 累積最低價 × (1 + 停損幅度)
trailing_stop_short = cumulative_low * (1 + stop_loss_percentage)

print("空頭追蹤停損價格計算：")
short_comparison_df = pd.DataFrame({
    '收盤價': tsmc_prices,
    '累積最低價': cumulative_low,
    '停損價（110%）': trailing_stop_short.round(2)
})
print(short_comparison_df.head(10))
print("\n" + "="*70 + "\n")

# 空頭停損觸發條件：收盤價 >= 停損價
stop_triggered_short = tsmc_prices >= trailing_stop_short

print("停損觸發檢查（空頭部位）：")
print(f"觸發停損天數：{stop_triggered_short.sum()}")

if stop_triggered_short.any():
    first_trigger_short = stop_triggered_short.idxmax()
    trigger_price_short = tsmc_prices.loc[first_trigger_short]
    stop_price_short = trailing_stop_short.loc[first_trigger_short]
    
    print(f"\n第一次觸發停損日期：{first_trigger_short.strftime('%Y-%m-%d')}")
    print(f"當日收盤價：{trigger_price_short:.2f} 元")
    print(f"停損價格：{stop_price_short:.2f} 元")
    
    # 計算做空的虧損（價格上漲則虧損）
    initial_short_price = tsmc_prices.iloc[0]
    short_return = ((initial_short_price - trigger_price_short) / 
                    initial_short_price * 100)
    print(f"做空報酬率：{short_return:.2f}%")

print("\n" + "="*70 + "\n")

# ============================================================================
# 範例三：動態調整停損幅度
# ============================================================================

def calculate_dynamic_trailing_stop(
    prices: pd.Series,
    initial_stop_pct: float = 0.10,
    tighten_after_days: int = 20,
    tightened_stop_pct: float = 0.05
) -> pd.Series:
    """
    動態調整追蹤停損幅度
    持有超過一定天數後，縮小停損幅度以保護更多獲利
    
    Parameters:
    -----------
    prices : pd.Series
        價格序列
    initial_stop_pct : float
        初始停損幅度（例如 0.10 代表 10%）
    tighten_after_days : int
        幾天後開始縮緊停損
    tightened_stop_pct : float
        縮緊後的停損幅度
    
    Returns:
    --------
    pd.Series
        動態調整的停損價格序列
    """
    cumulative_high = prices.cummax()
    
    # 建立停損幅度序列
    stop_percentages = pd.Series(
        [initial_stop_pct] * len(prices),
        index=prices.index
    )
    
    # 持有超過指定天數後，縮小停損幅度
    if len(prices) > tighten_after_days:
        stop_percentages.iloc[tighten_after_days:] = tightened_stop_pct
    
    # 計算動態停損價格
    protection_ratios = 1 - stop_percentages
    trailing_stops = cumulative_high * protection_ratios
    
    return trailing_stops

# 應用動態停損策略
dynamic_stops = calculate_dynamic_trailing_stop(
    tsmc_prices,
    initial_stop_pct=0.10,
    tighten_after_days=20,
    tightened_stop_pct=0.05
)

print("動態調整追蹤停損策略：")
dynamic_comparison = pd.DataFrame({
    '收盤價': tsmc_prices,
    '固定停損（10%）': trailing_stop_long.round(2),
    '動態停損': dynamic_stops.round(2)
})
print(dynamic_comparison.iloc[18:23])  # 顯示轉換點前後
print("\n" + "="*70 + "\n")

# ============================================================================
# 範例四：整合管道操作的完整交易策略
# ============================================================================

def load_price_data(symbol: str) -> pd.Series:
    """模擬載入股價資料"""
    # 實務上這裡會從資料庫或 API 讀取真實資料
    return tsmc_prices

def calculate_trailing_stop(
    prices: pd.Series,
    stop_percentage: float = 0.10,
    position: str = 'long'
) -> pd.Series:
    """計算追蹤停損價格"""
    if position == 'long':
        cumulative_extreme = prices.cummax()
        stop_prices = cumulative_extreme * (1 - stop_percentage)
    else:  # short
        cumulative_extreme = prices.cummin()
        stop_prices = cumulative_extreme * (1 + stop_percentage)
    
    return stop_prices

def check_stop_triggered(
    prices: pd.Series,
    stop_prices: pd.Series,
    position: str = 'long'
) -> pd.DataFrame:
    """檢查停損觸發狀況"""
    if position == 'long':
        triggered = prices <= stop_prices
    else:  # short
        triggered = prices >= stop_prices
    
    results = pd.DataFrame({
        '價格': prices,
        '停損價': stop_prices.round(2),
        '觸發': triggered
    })
    
    return results

# 使用管道式設計的交易策略分析
print("完整交易策略分析（管道式）：")
analysis_results = check_stop_triggered(
    tsmc_prices,
    calculate_trailing_stop(tsmc_prices, stop_percentage=0.08),
    position='long'
)

print(analysis_results.head(10))
print(f"\n總觸發次數：{analysis_results['觸發'].sum()}")

實務應用中，追蹤停損策略往往需要根據市場狀況動態調整。在趨勢明確的多頭市場中，可以使用較寬鬆的停損幅度（例如 15%-20%），給予價格更大的波動空間。在盤整或下跌趨勢中，則應該收緊停損幅度（例如 5%-8%），更積極地保護資本。台灣投資人可以根據技術指標、波動率指標或市場情緒指標動態調整停損參數，建立更靈活的風險管理機制。

停損觸發時點的分析同樣重要。idxmax 與 idxmin 方法能夠快速找出布林序列中第一個 True 值的位置，這在追蹤停損場景中對應著第一次觸及停損價的日期。透過這個資訊，我們可以計算持有期間、評估策略績效、分析停損頻率等關鍵指標。在台灣的量化交易系統中，這些分析結果會被記錄到交易日誌，供後續的策略最佳化與回測驗證使用。

Pandas 的進階資料處理技術為資料科學工作者提供了強大的工具集。管道操作讓複雜的資料轉換流程變得清晰易懂，Top N 分析提供了高效的極值篩選能力，累積極值方法則在金融應用中展現了獨特價值。對於台灣的資料科學從業人員而言，精通這些技術不僅能夠提升日常工作效率，更能夠在金融分析、商業智慧、運動數據分析等多個領域中創造實際價值。隨著資料量持續增長與分析需求日益複雜，這些看似簡單的方法往往能在關鍵時刻發揮意想不到的作用，值得每位資料分析師深入學習與靈活運用。