前言
當代公共衛生面臨的挑戰日益複雜,從傳染病防治、慢性病管理到醫療資源配置,都需要更精準、更即時的決策支援。資料科學的興起為這些挑戰提供了創新的解決途徑。透過機器學習演算法,我們能夠從龐大的健康資料中發現疾病傳播模式。利用預測分析技術,醫療機構可以提前規劃資源配置,避免醫療量能不足的困境。大數據分析則讓政策制定者能夠基於實證,設計更有效的公共衛生介入措施。
台灣擁有完善的全民健保制度與高品質的醫療資訊系統,累積了豐富的健康資料。如何善用這些資料資產,結合先進的資料科學技術,提升公共衛生服務品質,是我們這一代公衛與資訊專業人員的共同使命。本文將從學術與實務兩個面向,探討資料科學如何驅動公共衛生創新,並透過跨產業的成功案例,展示資料驅動決策的實踐路徑。
資料科學與公共衛生的學術先驅
語義網路與智慧資料系統的開拓者
在資料科學的發展歷程中,如何讓機器理解資料的語義關聯,一直是核心挑戰之一。Dean Allemang 在語義網路（Semantic Web）與連結資料（Linked Data）領域的開創性工作,為建構智慧型醫療資料系統奠定了重要基礎。語義網路技術能夠將分散在不同醫療機構的健康資料,透過標準化的本體論（Ontology）連結起來,讓電腦系統能夠理解疾病、症狀、藥物之間的複雜關聯。
這種技術在公共衛生監測系統中具有巨大潛力。舉例來說,當某地區出現不尋常的症狀通報模式時,語義網路系統可以自動搜尋相關的疾病知識庫,協助衛生單位快速判斷是否有新興傳染病的風險。在藥物警戒（Pharmacovigilance）領域,語義技術能夠整合來自不同來源的藥物不良反應資料,及早發現潛在的用藥安全問題。
Kirk Borne 作為資料科學教育的倡導者,長期致力於推廣資料素養（Data Literacy）的重要性。他強調,在資料爆炸的時代,不僅資料科學家需要具備分析能力,醫療從業人員、公衛決策者,甚至一般民眾,都應該培養基本的資料判讀能力。只有當整個社會都具備資料思維,資料驅動的公共衛生決策才能真正落地實現。
經濟學與發展研究的數據革命
諾貝爾經濟學獎得主 Esther Duflo 與 Abhijit Banerjee 的研究團隊,透過嚴謹的隨機對照試驗（Randomized Controlled Trials, RCT）方法,驗證各種貧困干預措施的實際效果。這種實證取向的研究典範,對公共衛生政策評估產生了深遠影響。傳統上,公衛介入措施的成效往往難以量化評估,政策制定多半仰賴專家經驗或理論推導。Duflo 等人證明,透過精心設計的實驗與嚴謹的資料分析,我們可以科學地評估不同政策選項的成本效益。
Michael Kremer 在疫苗接種誘因機制的研究,展示了如何運用經濟學理論與實驗資料,設計更有效的公共衛生方案。他發現,提供小額獎勵能夠顯著提高疫苗接種率,這個洞見已經被許多發展中國家採納,成為兒童免疫計畫的重要策略。台灣在推動流感疫苗接種或是慢性病篩檢時,也可以借鑑這些以數據為基礎的誘因設計經驗。
Emily Oster 將經濟學的資料分析方法應用於親子健康議題,協助父母在面對懷孕、育兒等人生重要決策時,能夠基於可靠的數據證據,而非僅憑道聽途說做選擇。她的著作《懷孕指南》（Expecting Better）系統性地回顧了孕期各種建議的科學證據,澄清了許多常見的迷思。這種將複雜的醫學文獻轉化為易懂決策建議的能力,正是資料科學在個人健康管理領域的重要應用。
全球健康倡議的資料驅動實踐
Bill Gates 透過比爾與梅琳達蓋茲基金會（Bill & Melinda Gates Foundation），將資料驅動方法應用於全球健康挑戰。基金會資助的研究專案都強調嚴謹的資料收集與成效評估,確保每一筆投資都能產生可衡量的健康改善。例如,在瘧疾防治方面,基金會支持開發即時疫情監測系統,運用衛星影像資料預測蚊蟲孳生地,並透過機器學習演算法優化蚊帳發放策略。
Jeffrey Sachs 在可持續發展目標（Sustainable Development Goals, SDGs）的推動過程中,建立了完整的指標監測框架,讓各國能夠追蹤健康、教育、環境等面向的進展。這些指標系統產生的資料,成為政策調整與資源配置的重要依據。台灣在參與國際衛生合作時,也應該善用資料分析能力,展現我們在精準公衛方面的專業實力。
Christopher Murray 領導的全球疾病負擔研究（Global Burden of Disease Study），建立了跨國比較的健康指標體系,讓我們能夠量化不同疾病對人群健康的影響。這項研究產生的失能調整生命年（Disability-Adjusted Life Years, DALYs）等指標,已經成為各國衛生政策規劃的標準工具。透過這些標準化的資料,政策制定者可以更客觀地評估應該優先處理哪些健康問題。
Steven Levitt 與 Stephen Dubner 在《蘋果橘子經濟學》（Freakonomics）中展示的分析思維,提醒我們資料背後可能隱藏著違反直覺的因果關係。這種批判性的資料思考方式,對於避免公衛政策的意外後果至關重要。例如,單純增加醫療資源投入,並不一定能改善健康結果,關鍵在於資源如何分配、誘因如何設計。
資料基礎設施的技術先驅
關聯式資料庫的革命性創新
Edgar F. Codd 在 1970 年提出的關聯式資料模型（Relational Database Model），徹底改變了資料管理的方式。在此之前,資料庫系統多半依賴複雜的指標結構,資料的存取與維護都需要專業程式設計師介入。關聯式模型以數學集合理論為基礎,透過簡潔的表格結構與 SQL 查詢語言,讓非技術背景的使用者也能夠操作資料庫。
這項創新對醫療資訊系統的發展具有決定性影響。現代的電子病歷系統、健保申報系統、疾病監測系統,都建構在關聯式資料庫之上。台灣的全民健保資料庫,儲存了超過兩千三百萬人的就醫紀錄,正是仰賴關聯式資料庫技術,才能支援大規模的查詢與分析作業。
Patricia Selinger 在查詢優化（Query Optimization）領域的貢獻,讓資料庫系統能夠自動選擇最有效率的查詢執行計畫。當公衛研究人員需要從龐大的健保資料庫中,擷取特定疾病患者的用藥資訊時,查詢優化器會分析各種可能的執行路徑,選擇最快速的方案。這種自動優化能力,大幅降低了資料分析的技術門檻。
Jim Gray 對資料庫交易處理（Transaction Processing）與分散式運算的研究,確保了資料的一致性與可靠性。在醫療場景中,交易的原子性（Atomicity）與隔離性（Isolation）至關重要。想像一個情境:當醫師開立處方的同時,藥師正在查詢藥物庫存,如果沒有適當的交易管理機制,可能導致重複用藥或庫存錯誤的問題。
Peter Chen 開發的實體關係模型（Entity-Relationship Model）提供了直觀的資料庫設計工具。透過實體、屬性、關聯等概念,資料庫設計者可以將複雜的業務需求轉化為清晰的資料結構。在設計疫情監測系統時,我們可以用實體關係圖來表達患者、醫療機構、檢驗結果、接觸史之間的複雜關聯,確保資料庫能夠支援各種分析需求。
import sqlite3
import pandas as pd
from datetime import datetime, timedelta
from typing import List, Dict, Tuple
import random
class PublicHealthDatabase:
"""
公共衛生資料庫管理系統
示範關聯式資料庫在疫情監測中的應用
"""
def __init__(self, db_path: str = "public_health.db"):
"""
初始化資料庫連線
Args:
db_path: 資料庫檔案路徑
"""
# 建立資料庫連線
self.conn = sqlite3.connect(db_path)
self.cursor = self.conn.cursor()
# 建立資料表結構
self._create_tables()
def _create_tables(self):
"""
建立公共衛生監測所需的資料表
實體關係設計:
- 患者 (Patient): 儲存基本人口學資料
- 就診記錄 (Visit): 記錄每次就醫資訊
- 診斷 (Diagnosis): 記錄疾病診斷
- 檢驗結果 (LabResult): 儲存實驗室檢驗資料
"""
# 患者基本資料表
self.cursor.execute("""
CREATE TABLE IF NOT EXISTS patients (
patient_id INTEGER PRIMARY KEY AUTOINCREMENT,
age INTEGER NOT NULL,
gender TEXT CHECK(gender IN ('M', 'F', 'O')),
district TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
# 就診記錄表
self.cursor.execute("""
CREATE TABLE IF NOT EXISTS visits (
visit_id INTEGER PRIMARY KEY AUTOINCREMENT,
patient_id INTEGER NOT NULL,
visit_date DATE NOT NULL,
hospital_id TEXT NOT NULL,
symptoms TEXT,
FOREIGN KEY (patient_id) REFERENCES patients(patient_id)
)
""")
# 診斷記錄表
self.cursor.execute("""
CREATE TABLE IF NOT EXISTS diagnoses (
diagnosis_id INTEGER PRIMARY KEY AUTOINCREMENT,
visit_id INTEGER NOT NULL,
icd_code TEXT NOT NULL,
disease_name TEXT NOT NULL,
severity TEXT CHECK(severity IN ('mild', 'moderate', 'severe')),
FOREIGN KEY (visit_id) REFERENCES visits(visit_id)
)
""")
# 檢驗結果表
self.cursor.execute("""
CREATE TABLE IF NOT EXISTS lab_results (
result_id INTEGER PRIMARY KEY AUTOINCREMENT,
visit_id INTEGER NOT NULL,
test_type TEXT NOT NULL,
result TEXT NOT NULL,
test_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (visit_id) REFERENCES visits(visit_id)
)
""")
# 提交變更
self.conn.commit()
print("資料表建立完成")
def insert_patient(self, age: int, gender: str, district: str) -> int:
"""
新增患者資料
Args:
age: 年齡
gender: 性別 (M/F/O)
district: 行政區
Returns:
新增患者的 ID
"""
self.cursor.execute("""
INSERT INTO patients (age, gender, district)
VALUES (?, ?, ?)
""", (age, gender, district))
self.conn.commit()
return self.cursor.lastrowid
def insert_visit(
self,
patient_id: int,
visit_date: str,
hospital_id: str,
symptoms: str
) -> int:
"""
新增就診記錄
Args:
patient_id: 患者 ID
visit_date: 就診日期 (YYYY-MM-DD)
hospital_id: 醫療機構代碼
symptoms: 症狀描述
Returns:
新增就診記錄的 ID
"""
self.cursor.execute("""
INSERT INTO visits (patient_id, visit_date, hospital_id, symptoms)
VALUES (?, ?, ?, ?)
""", (patient_id, visit_date, hospital_id, symptoms))
self.conn.commit()
return self.cursor.lastrowid
def insert_diagnosis(
self,
visit_id: int,
icd_code: str,
disease_name: str,
severity: str
) -> int:
"""
新增診斷記錄
Args:
visit_id: 就診記錄 ID
icd_code: ICD 疾病分類碼
disease_name: 疾病名稱
severity: 嚴重程度 (mild/moderate/severe)
Returns:
新增診斷記錄的 ID
"""
self.cursor.execute("""
INSERT INTO diagnoses (visit_id, icd_code, disease_name, severity)
VALUES (?, ?, ?, ?)
""", (visit_id, icd_code, disease_name, severity))
self.conn.commit()
return self.cursor.lastrowid
def get_disease_outbreak_alert(
self,
disease_icd: str,
days: int = 7,
threshold: int = 10
) -> pd.DataFrame:
"""
疾病爆發預警系統
監測特定疾病在近期的病例數變化
Args:
disease_icd: 疾病 ICD 代碼
days: 監測天數
threshold: 病例數閾值
Returns:
包含每日病例數的 DataFrame
"""
# 計算監測起始日期
end_date = datetime.now().date()
start_date = end_date - timedelta(days=days)
# 查詢指定期間內的病例數
query = """
SELECT
v.visit_date,
COUNT(DISTINCT v.patient_id) as case_count,
d.disease_name
FROM visits v
JOIN diagnoses d ON v.visit_id = d.visit_id
WHERE d.icd_code = ?
AND v.visit_date BETWEEN ? AND ?
GROUP BY v.visit_date, d.disease_name
ORDER BY v.visit_date
"""
# 執行查詢
df = pd.read_sql_query(
query,
self.conn,
params=(disease_icd, start_date, end_date)
)
# 判斷是否超過閾值
if len(df) > 0 and df['case_count'].max() >= threshold:
print(f"⚠️ 警告: {df['disease_name'].iloc[0]} 病例數超過閾值!")
print(f"最高單日病例數: {df['case_count'].max()}")
return df
def analyze_disease_distribution_by_district(
self,
disease_icd: str
) -> pd.DataFrame:
"""
分析疾病的地理分布
協助衛生單位識別高風險區域
Args:
disease_icd: 疾病 ICD 代碼
Returns:
各行政區的病例數統計
"""
query = """
SELECT
p.district,
COUNT(DISTINCT v.patient_id) as case_count,
d.disease_name
FROM patients p
JOIN visits v ON p.patient_id = v.patient_id
JOIN diagnoses d ON v.visit_id = d.visit_id
WHERE d.icd_code = ?
GROUP BY p.district, d.disease_name
ORDER BY case_count DESC
"""
df = pd.read_sql_query(query, self.conn, params=(disease_icd,))
return df
def get_age_group_vulnerability(
self,
disease_icd: str
) -> pd.DataFrame:
"""
分析不同年齡層對特定疾病的脆弱性
Args:
disease_icd: 疾病 ICD 代碼
Returns:
年齡層病例數與嚴重度統計
"""
query = """
SELECT
CASE
WHEN p.age < 18 THEN '0-17歲'
WHEN p.age < 45 THEN '18-44歲'
WHEN p.age < 65 THEN '45-64歲'
ELSE '65歲以上'
END as age_group,
COUNT(DISTINCT v.patient_id) as case_count,
SUM(CASE WHEN d.severity = 'severe' THEN 1 ELSE 0 END) as severe_cases,
ROUND(
100.0 * SUM(CASE WHEN d.severity = 'severe' THEN 1 ELSE 0 END) /
COUNT(*),
2
) as severe_percentage
FROM patients p
JOIN visits v ON p.patient_id = v.patient_id
JOIN diagnoses d ON v.visit_id = d.visit_id
WHERE d.icd_code = ?
GROUP BY age_group
ORDER BY
CASE age_group
WHEN '0-17歲' THEN 1
WHEN '18-44歲' THEN 2
WHEN '45-64歲' THEN 3
ELSE 4
END
"""
df = pd.read_sql_query(query, self.conn, params=(disease_icd,))
return df
def close(self):
"""關閉資料庫連線"""
self.conn.close()
print("資料庫連線已關閉")
# 使用範例與模擬資料生成
if __name__ == "__main__":
# 建立資料庫實例
db = PublicHealthDatabase("epidemic_surveillance.db")
# 模擬資料生成 - 新增患者與就診記錄
print("=" * 60)
print("正在生成模擬疫情資料...")
print("=" * 60)
# 台北市各行政區
districts = ['中正區', '大同區', '中山區', '松山區', '大安區',
'萬華區', '信義區', '士林區', '北投區', '內湖區',
'南港區', '文山區']
# 模擬流感疫情資料
flu_symptoms = [
'發燒、咳嗽、喉嚨痛',
'發燒、頭痛、肌肉痠痛',
'咳嗽、流鼻水、疲倦',
'高燒、畏寒、全身無力'
]
# 生成 100 筆模擬資料
for i in range(100):
# 隨機生成患者資料
age = random.randint(5, 85)
gender = random.choice(['M', 'F'])
district = random.choice(districts)
# 新增患者
patient_id = db.insert_patient(age, gender, district)
# 生成就診日期 (最近 14 天內)
days_ago = random.randint(0, 14)
visit_date = (datetime.now() - timedelta(days=days_ago)).strftime('%Y-%m-%d')
# 新增就診記錄
hospital_id = f"H{random.randint(1, 10):03d}"
symptoms = random.choice(flu_symptoms)
visit_id = db.insert_visit(patient_id, visit_date, hospital_id, symptoms)
# 新增診斷記錄 (流感)
severity = random.choices(
['mild', 'moderate', 'severe'],
weights=[0.6, 0.3, 0.1] # 60% 輕症, 30% 中症, 10% 重症
)[0]
db.insert_diagnosis(visit_id, 'J10', '流感', severity)
print(f"✅ 已生成 100 筆模擬疫情資料\n")
# 執行疾病爆發預警分析
print("=" * 60)
print("疾病爆發預警分析")
print("=" * 60)
outbreak_df = db.get_disease_outbreak_alert(
disease_icd='J10',
days=14,
threshold=5
)
print("\n近 14 天流感病例趨勢:")
print(outbreak_df.to_string(index=False))
# 分析地理分布
print("\n" + "=" * 60)
print("疾病地理分布分析")
print("=" * 60)
geo_df = db.analyze_disease_distribution_by_district('J10')
print("\n各行政區流感病例數:")
print(geo_df.to_string(index=False))
# 分析年齡層脆弱性
print("\n" + "=" * 60)
print("年齡層脆弱性分析")
print("=" * 60)
age_df = db.get_age_group_vulnerability('J10')
print("\n不同年齡層流感病例與嚴重度:")
print(age_df.to_string(index=False))
# 關閉資料庫連線
db.close()
大數據時代的資料架構創新
James Dixon 提出的「資料湖」（Data Lake）概念,回應了傳統資料倉儲在處理非結構化資料時的局限。資料倉儲通常要求資料在載入前就定義好嚴格的結構（Schema-on-Write），這對於醫療影像、基因序列、臨床筆記等多元資料類型來說過於僵化。資料湖採用「寫入時不定義結構,讀取時才定義」（Schema-on-Read）的彈性架構,讓研究人員能夠先收集資料,之後再根據分析需求定義結構。
在精準醫療的脈絡下,資料湖技術尤其重要。整合基因體資料、電子病歷、生活型態資訊、環境暴露資料,需要能夠處理多種資料格式的基礎架構。台灣的生醫資料庫若能採用資料湖架構,將更有利於跨領域研究團隊進行創新分析。
Doug Cutting 開發的 Hadoop 開源專案,讓大規模分散式資料處理技術不再是科技巨頭的專利。Hadoop 的核心概念是將計算任務分散到多台電腦上平行執行,大幅縮短資料處理時間。在流行病學研究中,當我們需要分析數百萬筆就醫記錄,尋找特定藥物與不良事件的關聯時,分散式運算能力能夠將原本需要數天的分析縮短到數小時。
Jeff Dean 在 Google 開發的 MapReduce 與 Bigtable 系統,展示了如何在數十萬台伺服器上協調複雜的資料處理工作。這些技術後來啟發了整個大數據生態系的發展,包括 Apache Spark、Apache Flink 等新一代資料處理框架。台灣的醫療機構雖然資源有限,但透過雲端服務,也能夠取用這些強大的運算能力。
深度學習的技術革命
循環神經網路的突破
Jürgen Schmidhuber 與 Sepp Hochreiter 共同開發的長短期記憶網路（Long Short-Term Memory, LSTM），解決了傳統循環神經網路（Recurrent Neural Network, RNN）在處理長序列資料時的梯度消失問題。LSTM 的創新在於引入記憶細胞（Memory Cell）與閘門機制（Gate Mechanism），讓網路能夠選擇性地記住或遺忘資訊。
這項技術在醫療時序資料分析中展現出色表現。患者的生命徵象監測資料,如心電圖、血壓變化、血糖波動,都是典型的時間序列。LSTM 能夠學習這些資料的長期相依模式,預測患者是否即將發生敗血症、心律不整或低血糖等危急狀況。台灣的加護病房若能部署 LSTM 預警系統,有機會更早介入救治,降低重症死亡率。
Alex Graves 將 LSTM 應用於手寫辨識,開創了序列到序列學習（Sequence-to-Sequence Learning）的新範式。在醫療領域,這項技術可以用來辨識醫師的手寫處方籤,降低因字跡潦草造成的給藥錯誤。更進階的應用包括從臨床筆記中擷取結構化資訊,自動產生診斷編碼,減輕醫療人員的行政負擔。
深度學習的學術領袖
Yoshua Bengio、Geoffrey Hinton、Yann LeCun 三位深度學習的先驅,因其對神經網路的突破性貢獻共同獲得 2018 年圖靈獎。他們在類神經網路的理論基礎、訓練演算法、架構設計等方面的研究,奠定了當代人工智慧的技術基石。
Ian Goodfellow 發明的生成對抗網路（Generative Adversarial Network, GAN），開啟了生成式人工智慧的新紀元。GAN 由兩個神經網路組成:生成器試圖產生逼真的假資料,判別器則試圖分辨真假。兩者在對抗過程中不斷進步,最終生成器能夠產生難以分辨的合成資料。在醫療研究中,GAN 可以生成合成病歷資料,讓研究人員在不洩露真實患者隱私的前提下,測試演算法的效能。
自然語言處理的醫療應用
Christopher Manning 領導的史丹佛自然語言處理團隊,開發了多項影響深遠的開源工具。Stanford CoreNLP 提供了詞性標註、命名實體辨識、依存句法分析等基礎功能,讓電腦能夠理解人類語言的結構。在處理中文醫療文本時,這些技術需要針對醫學專業術語進行調適,但基本原理是相通的。
台灣的電子病歷系統累積了龐大的中文臨床筆記,這些非結構化文本蘊含豐富的臨床資訊,卻難以直接用於量化分析。自然語言處理技術能夠從這些筆記中擷取症狀、檢查結果、用藥資訊、治療反應等結構化資料,大幅提升資料的可用性。
Yoav Goldberg 與 Sebastian Ruder 在神經網路應用於 NLP 的研究,特別關注遷移學習（Transfer Learning）技術。預訓練語言模型如 BERT、GPT 系列,在大規模文本上學習語言的通用表徵,然後透過少量標註資料就能適應特定任務。這對資源有限的醫療 NLP 專案特別有價值,因為標註醫療文本需要專業醫師參與,成本高昂。透過遷移學習,我們可以用較少的標註資料,達到良好的效能。
import numpy as np
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from tensorflow import keras
from tensorflow.keras import layers
from typing import Tuple
import matplotlib.pyplot as plt
# 設定中文字型
plt.rcParams['font.sans-serif'] = ['Microsoft JhengHei']
plt.rcParams['axes.unicode_minus'] = False
class SepsisEarlyWarningSystem:
"""
敗血症早期預警系統
使用 LSTM 模型分析患者生命徵象時序資料,預測敗血症風險
"""
def __init__(self, sequence_length: int = 12, n_features: int = 5):
"""
初始化預警系統
Args:
sequence_length: 時間序列長度 (小時數)
n_features: 特徵數量 (生命徵象指標數)
"""
self.sequence_length = sequence_length
self.n_features = n_features
self.model = None
self.scaler = StandardScaler()
def build_lstm_model(self) -> keras.Model:
"""
建構 LSTM 預測模型
模型架構:
1. LSTM 層: 學習時序依賴關係
2. Dropout 層: 防止過擬合
3. Dense 層: 最終預測輸出
Returns:
編譯完成的 Keras 模型
"""
# 定義模型架構
model = keras.Sequential([
# 第一層 LSTM: 128 個記憶單元,返回完整序列
layers.LSTM(
128,
return_sequences=True,
input_shape=(self.sequence_length, self.n_features),
name='lstm_layer_1'
),
# Dropout 層: 隨機丟棄 20% 的神經元,防止過擬合
layers.Dropout(0.2, name='dropout_1'),
# 第二層 LSTM: 64 個記憶單元
layers.LSTM(64, return_sequences=False, name='lstm_layer_2'),
# Dropout 層
layers.Dropout(0.2, name='dropout_2'),
# 全連接層: 32 個神經元,使用 ReLU 激活函數
layers.Dense(32, activation='relu', name='dense_1'),
# 輸出層: 使用 Sigmoid 激活函數,輸出 0-1 之間的機率值
layers.Dense(1, activation='sigmoid', name='output')
])
# 編譯模型
# 使用 Adam 優化器,二元交叉熵損失函數
model.compile(
optimizer=keras.optimizers.Adam(learning_rate=0.001),
loss='binary_crossentropy',
metrics=[
'accuracy',
keras.metrics.AUC(name='auc'), # ROC-AUC 評估指標
keras.metrics.Precision(name='precision'),
keras.metrics.Recall(name='recall')
]
)
return model
def generate_simulated_vitals_data(
self,
n_patients: int = 1000,
sepsis_rate: float = 0.15
) -> Tuple[np.ndarray, np.ndarray]:
"""
生成模擬的生命徵象時序資料
模擬特徵:
1. 心率 (bpm)
2. 收縮壓 (mmHg)
3. 體溫 (°C)
4. 呼吸速率 (次/分)
5. 血氧飽和度 (%)
Args:
n_patients: 患者數量
sepsis_rate: 敗血症發生率
Returns:
X: 形狀為 (n_patients, sequence_length, n_features) 的特徵陣列
y: 形狀為 (n_patients,) 的標籤陣列
"""
np.random.seed(42)
# 正常範圍的生命徵象基準值
normal_vitals = {
'heart_rate': (70, 10), # 平均值, 標準差
'systolic_bp': (120, 12),
'temperature': (36.8, 0.3),
'respiratory_rate': (16, 3),
'spo2': (98, 1)
}
# 敗血症患者的生命徵象異常模式
sepsis_vitals = {
'heart_rate': (110, 15), # 心率升高
'systolic_bp': (95, 15), # 血壓下降
'temperature': (38.5, 0.8), # 體溫升高
'respiratory_rate': (24, 5), # 呼吸加速
'spo2': (94, 3) # 血氧下降
}
X = np.zeros((n_patients, self.sequence_length, self.n_features))
y = np.zeros(n_patients)
# 確定敗血症患者數量
n_sepsis = int(n_patients * sepsis_rate)
sepsis_indices = np.random.choice(n_patients, n_sepsis, replace=False)
y[sepsis_indices] = 1
for i in range(n_patients):
# 根據是否為敗血症患者選擇參數
if i in sepsis_indices:
vitals = sepsis_vitals
# 敗血症患者的生命徵象隨時間惡化
deterioration = np.linspace(0, 1, self.sequence_length)
else:
vitals = normal_vitals
deterioration = np.zeros(self.sequence_length)
# 生成每個時間點的生命徵象
for t in range(self.sequence_length):
# 心率
X[i, t, 0] = np.random.normal(
vitals['heart_rate'][0] + deterioration[t] * 20,
vitals['heart_rate'][1]
)
# 收縮壓
X[i, t, 1] = np.random.normal(
vitals['systolic_bp'][0] - deterioration[t] * 15,
vitals['systolic_bp'][1]
)
# 體溫
X[i, t, 2] = np.random.normal(
vitals['temperature'][0] + deterioration[t] * 1.5,
vitals['temperature'][1]
)
# 呼吸速率
X[i, t, 3] = np.random.normal(
vitals['respiratory_rate'][0] + deterioration[t] * 8,
vitals['respiratory_rate'][1]
)
# 血氧飽和度
X[i, t, 4] = np.random.normal(
vitals['spo2'][0] - deterioration[t] * 6,
vitals['spo2'][1]
)
return X, y
def train(
self,
X_train: np.ndarray,
y_train: np.ndarray,
X_val: np.ndarray,
y_val: np.ndarray,
epochs: int = 50,
batch_size: int = 32
) -> keras.callbacks.History:
"""
訓練 LSTM 模型
Args:
X_train: 訓練集特徵
y_train: 訓練集標籤
X_val: 驗證集特徵
y_val: 驗證集標籤
epochs: 訓練週期數
batch_size: 批次大小
Returns:
訓練歷史記錄
"""
# 標準化特徵
# 將三維陣列重塑為二維進行標準化
n_samples, n_timesteps, n_features = X_train.shape
X_train_reshaped = X_train.reshape(-1, n_features)
X_val_reshaped = X_val.reshape(-1, n_features)
# 擬合標準化器並轉換訓練資料
X_train_scaled = self.scaler.fit_transform(X_train_reshaped)
X_val_scaled = self.scaler.transform(X_val_reshaped)
# 重塑回三維陣列
X_train_scaled = X_train_scaled.reshape(n_samples, n_timesteps, n_features)
X_val_scaled = X_val_scaled.reshape(X_val.shape[0], n_timesteps, n_features)
# 建構模型
self.model = self.build_lstm_model()
# 設定早停機制,避免過擬合
early_stopping = keras.callbacks.EarlyStopping(
monitor='val_loss',
patience=10,
restore_best_weights=True,
verbose=1
)
# 訓練模型
print("開始訓練 LSTM 模型...")
history = self.model.fit(
X_train_scaled,
y_train,
validation_data=(X_val_scaled, y_val),
epochs=epochs,
batch_size=batch_size,
callbacks=[early_stopping],
verbose=1
)
return history
def predict(self, X: np.ndarray) -> np.ndarray:
"""
預測敗血症風險
Args:
X: 生命徵象時序資料
Returns:
敗血症風險機率 (0-1)
"""
# 標準化特徵
n_samples, n_timesteps, n_features = X.shape
X_reshaped = X.reshape(-1, n_features)
X_scaled = self.scaler.transform(X_reshaped)
X_scaled = X_scaled.reshape(n_samples, n_timesteps, n_features)
# 進行預測
predictions = self.model.predict(X_scaled, verbose=0)
return predictions.flatten()
def evaluate(self, X_test: np.ndarray, y_test: np.ndarray):
"""
評估模型效能
Args:
X_test: 測試集特徵
y_test: 測試集標籤
"""
# 標準化特徵
n_samples, n_timesteps, n_features = X_test.shape
X_test_reshaped = X_test.reshape(-1, n_features)
X_test_scaled = self.scaler.transform(X_test_reshaped)
X_test_scaled = X_test_scaled.reshape(n_samples, n_timesteps, n_features)
# 評估模型
results = self.model.evaluate(X_test_scaled, y_test, verbose=0)
print("\n" + "=" * 60)
print("模型評估結果")
print("=" * 60)
print(f"損失值 (Loss): {results[0]:.4f}")
print(f"準確率 (Accuracy): {results[1]:.4f}")
print(f"AUC-ROC: {results[2]:.4f}")
print(f"精確率 (Precision): {results[3]:.4f}")
print(f"召回率 (Recall): {results[4]:.4f}")
# 計算 F1 分數
precision, recall = results[3], results[4]
f1_score = 2 * (precision * recall) / (precision + recall) if (precision + recall) > 0 else 0
print(f"F1 分數: {f1_score:.4f}")
def plot_training_history(self, history: keras.callbacks.History):
"""
視覺化訓練過程
Args:
history: 訓練歷史記錄
"""
fig, axes = plt.subplots(2, 2, figsize=(14, 10))
# 損失值曲線
axes[0, 0].plot(history.history['loss'], label='訓練集')
axes[0, 0].plot(history.history['val_loss'], label='驗證集')
axes[0, 0].set_title('模型損失值', fontsize=14)
axes[0, 0].set_xlabel('訓練週期', fontsize=12)
axes[0, 0].set_ylabel('損失值', fontsize=12)
axes[0, 0].legend()
axes[0, 0].grid(True, alpha=0.3)
# 準確率曲線
axes[0, 1].plot(history.history['accuracy'], label='訓練集')
axes[0, 1].plot(history.history['val_accuracy'], label='驗證集')
axes[0, 1].set_title('模型準確率', fontsize=14)
axes[0, 1].set_xlabel('訓練週期', fontsize=12)
axes[0, 1].set_ylabel('準確率', fontsize=12)
axes[0, 1].legend()
axes[0, 1].grid(True, alpha=0.3)
# AUC 曲線
axes[1, 0].plot(history.history['auc'], label='訓練集')
axes[1, 0].plot(history.history['val_auc'], label='驗證集')
axes[1, 0].set_title('AUC-ROC', fontsize=14)
axes[1, 0].set_xlabel('訓練週期', fontsize=12)
axes[0, 1].set_ylabel('AUC', fontsize=12)
axes[1, 0].legend()
axes[1, 0].grid(True, alpha=0.3)
# 精確率與召回率曲線
axes[1, 1].plot(history.history['precision'], label='精確率 (訓練)')
axes[1, 1].plot(history.history['val_precision'], label='精確率 (驗證)')
axes[1, 1].plot(history.history['recall'], label='召回率 (訓練)')
axes[1, 1].plot(history.history['val_recall'], label='召回率 (驗證)')
axes[1, 1].set_title('精確率與召回率', fontsize=14)
axes[1, 1].set_xlabel('訓練週期', fontsize=12)
axes[1, 1].set_ylabel('分數', fontsize=12)
axes[1, 1].legend()
axes[1, 1].grid(True, alpha=0.3)
plt.tight_layout()
plt.savefig('lstm_training_history.png', dpi=300, bbox_inches='tight')
print("\n訓練歷史圖表已儲存至 lstm_training_history.png")
# 使用範例
if __name__ == "__main__":
print("=" * 60)
print("敗血症早期預警系統 - LSTM 模型訓練")
print("=" * 60)
# 建立預警系統實例
ews = SepsisEarlyWarningSystem(sequence_length=12, n_features=5)
# 生成模擬資料
print("\n正在生成模擬生命徵象資料...")
X, y = ews.generate_simulated_vitals_data(n_patients=2000, sepsis_rate=0.15)
print(f"資料形狀: {X.shape}")
print(f"敗血症比例: {y.mean():.2%}")
# 分割資料集
X_train, X_temp, y_train, y_temp = train_test_split(
X, y, test_size=0.3, random_state=42, stratify=y
)
X_val, X_test, y_val, y_test = train_test_split(
X_temp, y_temp, test_size=0.5, random_state=42, stratify=y_temp
)
print(f"\n訓練集大小: {len(X_train)}")
print(f"驗證集大小: {len(X_val)}")
print(f"測試集大小: {len(X_test)}")
# 訓練模型
history = ews.train(
X_train, y_train,
X_val, y_val,
epochs=50,
batch_size=32
)
# 評估模型
ews.evaluate(X_test, y_test)
# 繪製訓練歷史
ews.plot_training_history(history)
# 示範預測
print("\n" + "=" * 60)
print("預測示範")
print("=" * 60)
# 隨機選取 5 個測試樣本
sample_indices = np.random.choice(len(X_test), 5, replace=False)
X_samples = X_test[sample_indices]
y_samples = y_test[sample_indices]
# 進行預測
predictions = ews.predict(X_samples)
print("\n患者敗血症風險預測:")
print("-" * 60)
for i, (pred, true) in enumerate(zip(predictions, y_samples)):
risk_level = "高風險" if pred > 0.5 else "低風險"
actual = "確診敗血症" if true == 1 else "未發生敗血症"
print(f"患者 {i+1}: 風險分數 = {pred:.4f} ({risk_level}) | 實際結果 = {actual}")
產業資料驅動決策的成功案例
Netflix 的內容策略革命
Netflix 從 DVD 郵寄服務轉型為全球串流媒體龍頭的過程,充分展現了資料驅動決策的威力。該公司累積了龐大的使用者行為資料,包括觀看時間、暫停位置、重複觀看、評分行為等細緻指標。透過分析這些資料,Netflix 不僅能夠個人化推薦內容,更能預測哪些類型的原創作品會受歡迎。
著名的《紙牌屋》影集,就是資料驅動內容決策的代表作。Netflix 分析發現,觀眾對政治題材、導演 David Fincher 的作品、演員 Kevin Spacey 的表演都有高度興趣。基於這些洞見,Netflix 決定投資製作《紙牌屋》,結果大獲成功,開啟了串流平台自製內容的新紀元。
這個案例對公共衛生傳播的啟示是,我們也可以運用資料分析,了解民眾對不同健康議題的關注度與偏好的訊息呈現方式。疫情期間,衛福部若能分析社群媒體上的互動資料,就能更精準地設計防疫宣導內容,提高民眾的接受度與遵從意願。
沃爾瑪的災難應變優化
沃爾瑪透過分析歷史銷售資料,發現了一個有趣的模式:在颶風來臨前,除了預期的飲用水與電池外,Pop-Tarts 草莓口味的銷量會大幅增加。這個洞見讓沃爾瑪能夠在颱風警報發布後,迅速調整庫存配置,確保門市不會缺貨。
這種預測性庫存管理的邏輯,完全適用於公共衛生的資源配置。當流感季節來臨時,醫療機構可以根據歷史就診資料,預測疫苗、抗病毒藥物、快篩試劑的需求量,提前備貨。在登革熱疫情方面,環保局可以分析氣候資料與過去的疫情模式,預測高風險區域,優先配置防治資源。
台灣經常面臨颱風、地震等天然災害,衛生單位若能建立類似的預測模型,根據災害類型與規模,預估醫療需求與藥品消耗,將能更有效地調度資源,減少災後的醫療量能不足問題。
UPS 的路線最佳化系統
UPS 開發的 ORION（On-Road Integrated Optimization and Navigation）系統,運用先進的演算法優化送貨路線。系統會考量包裹數量、交通狀況、客戶時間視窗等多重限制,計算出最有效率的路線。特別值得注意的是,ORION 會盡量減少左轉,因為左轉需要等待對向車流,增加油耗與時間成本。
這套系統為 UPS 每年節省數千萬公升的燃油與數百萬美元的成本。在公共衛生領域,類似的路線優化技術可以應用於行動醫療車的巡迴規劃、疫苗到府接種的路徑安排、救護車的派遣調度等情境。台灣的偏鄉醫療資源不足,若能透過優化排程,讓有限的醫療人力發揮最大效益,將能改善醫療可近性的問題。
星巴克的選址策略
星巴克運用地理資訊系統（Geographic Information System, GIS）與人口統計資料,科學化地評估潛在店址。分析因素包括人流量、人口密度、收入水平、競爭對手位置、交通可達性等。透過建立預測模型,星巴克能夠在開店前就相當準確地預估業績表現。
這種資料驅動的選址方法,對於醫療資源配置規劃同樣有價值。衛生局在規劃社區健康中心、疫苗接種站、心理諮商據點時,可以整合人口資料、疾病盛行率、交通網路等資訊,找出最能服務弱勢族群的設置地點。台灣的長照據點布建,若能採用類似的分析方法,將能更公平地分配資源,縮小城鄉差距。
Airbnb 的動態定價模型
Airbnb 開發的機器學習定價模型,會根據需求變化、當地活動、季節性、訂房趨勢等因素,動態調整房源價格建議。這不僅幫助房東優化收益,也讓旅客在淡季能夠享受更優惠的價格,提高整體平台的使用率。
動態定價的概念在醫療領域的應用雖然敏感,但並非全無可能。例如,牙醫診所在離峰時段提供折扣,鼓勵患者分散就診時間,減少擁擠。健檢中心可以根據預約狀況,動態調整價格,提高設備使用率。當然,這些應用必須審慎評估,確保不會影響醫療公平性。
更適合的應用場景是公共衛生服務的資源動態調配。當某地區流感疫苗施打率偏低時,可以增派醫療人員或延長服務時間。透過即時監測需求訊號,彈性調整資源配置,提高公共衛生服務的效率與覆蓋率。
Target 的預測分析爭議
Target 的資料科學團隊透過分析顧客購買行為,開發出預測懷孕的演算法。當顧客開始購買無香味乳液、維生素補充劑、特定款式的手提包時,模型會推測她可能懷孕了,並推播相關產品廣告。
這個案例在行銷界成為經典,但也引發隱私爭議。有位父親向 Target 抗議寄送嬰兒用品折價券給他的高中女兒,後來才發現女兒確實懷孕了。這凸顯了資料分析的倫理挑戰:即使預測準確,是否有權在當事人尚未公開前就採取行動?
公共衛生領域也面臨類似的倫理困境。透過分析健保就醫資料,我們能夠預測誰可能罹患某些疾病,但是否應該主動介入?如何在維護隱私與促進健康之間取得平衡?台灣在發展精準公衛的過程中,必須建立清楚的倫理指引,確保資料應用不會侵犯個人權益。
福特汽車的客戶導向設計
福特汽車整合客戶反饋、調查資料、實際駕駛行為,優化車輛設計。例如,分析發現皮卡車車主經常需要拖曳重物,福特便強化了 F-150 系列的拖曳能力與耐用性。透過資料了解客戶真實需求,而非僅憑工程師的主觀判斷,讓產品更貼近市場期待。
這種以使用者為中心的設計思維,在公共衛生服務改善上同樣重要。衛生單位常常基於專業判斷設計服務流程,卻忽略了民眾的實際需求與使用障礙。透過分析服務使用資料、收集回饋意見,可以發現流程中的痛點,進行優化。
例如,許多長者反映線上預約疫苗接種太複雜,若衛生局分析發現特定年齡層的線上預約完成率明顯偏低,就應該提供更友善的介面或加強電話預約管道。資料驅動的服務設計,能夠讓公衛資源真正發揮效益,服務到需要的人群。
洛杉磯的預測性警務
洛杉磯警察局運用機器學習演算法,分析歷史犯罪資料,預測未來可能發生犯罪的熱點區域與時段。系統會建議警力部署策略,讓有限的警力資源發揮最大嚇阻效果。實施後,某些地區的犯罪率確實下降了。
然而,這項計畫也引發激烈爭議。批評者指出,演算法可能強化既有的偏見,導致特定族群或社區被過度警務化。如果模型是基於歷史資料訓練,而歷史資料本身就反映了執法偏見,那麼演算法只會複製甚至放大這些偏見。
公共衛生的疾病熱點分析也可能面臨類似問題。若某社區因為醫療資源較好而有較高的疾病通報率,演算法可能誤判該區為高風險區,配置更多資源,形成「富者更富」的馬太效應。因此,在應用預測分析時,必須審慎檢視資料品質與演算法公平性,避免加劇健康不平等。
import numpy as np
import pandas as pd
from sklearn.cluster import DBSCAN
from sklearn.preprocessing import StandardScaler
import matplotlib.pyplot as plt
from datetime import datetime, timedelta
import folium
from typing import List, Tuple
# 設定中文字型
plt.rcParams['font.sans-serif'] = ['Microsoft JhengHei']
plt.rcParams['axes.unicode_minus'] = False
class DiseaseHotspotAnalyzer:
"""
疾病熱點分析系統
使用空間聚類演算法識別疾病爆發的地理熱點
"""
def __init__(self, eps_km: float = 1.0, min_samples: int = 5):
"""
初始化熱點分析系統
Args:
eps_km: DBSCAN 鄰域半徑 (公里)
min_samples: 形成聚類的最小樣本數
"""
self.eps_km = eps_km
self.min_samples = min_samples
self.scaler = StandardScaler()
def generate_disease_cases(
self,
n_cases: int = 500,
n_hotspots: int = 3
) -> pd.DataFrame:
"""
生成模擬的疾病案例資料
Args:
n_cases: 案例總數
n_hotspots: 熱點區域數量
Returns:
包含案例地理座標與時間的 DataFrame
"""
np.random.seed(42)
# 台北市中心座標
taipei_center = (25.0330, 121.5654)
# 生成熱點中心
hotspot_centers = []
for i in range(n_hotspots):
# 在台北市範圍內隨機生成熱點中心
lat = taipei_center[0] + np.random.uniform(-0.05, 0.05)
lon = taipei_center[1] + np.random.uniform(-0.05, 0.05)
hotspot_centers.append((lat, lon))
cases = []
# 70% 的案例集中在熱點區域
n_hotspot_cases = int(n_cases * 0.7)
for i in range(n_hotspot_cases):
# 隨機選擇一個熱點
center = hotspot_centers[np.random.choice(n_hotspots)]
# 在熱點中心周圍生成案例 (高斯分布)
lat = np.random.normal(center[0], 0.005) # 約 500 公尺標準差
lon = np.random.normal(center[1], 0.005)
# 生成案例時間 (最近 30 天內)
days_ago = np.random.randint(0, 30)
case_date = datetime.now() - timedelta(days=days_ago)
cases.append({
'case_id': i + 1,
'latitude': lat,
'longitude': lon,
'case_date': case_date,
'district': self._get_district(lat, lon)
})
# 30% 的案例隨機分布 (背景案例)
n_background_cases = n_cases - n_hotspot_cases
for i in range(n_background_cases):
lat = taipei_center[0] + np.random.uniform(-0.08, 0.08)
lon = taipei_center[1] + np.random.uniform(-0.08, 0.08)
days_ago = np.random.randint(0, 30)
case_date = datetime.now() - timedelta(days=days_ago)
cases.append({
'case_id': n_hotspot_cases + i + 1,
'latitude': lat,
'longitude': lon,
'case_date': case_date,
'district': self._get_district(lat, lon)
})
df = pd.DataFrame(cases)
return df
def _get_district(self, lat: float, lon: float) -> str:
"""
根據座標推估行政區 (簡化版)
Args:
lat: 緯度
lon: 經度
Returns:
行政區名稱
"""
# 這是簡化的區域劃分,實際應使用 GIS 系統
districts = ['中正區', '大同區', '中山區', '松山區', '大安區',
'萬華區', '信義區', '士林區', '北投區', '內湖區']
return np.random.choice(districts)
def identify_hotspots(
self,
df: pd.DataFrame,
recent_days: int = 7
) -> Tuple[pd.DataFrame, np.ndarray]:
"""
識別疾病熱點
Args:
df: 案例資料 DataFrame
recent_days: 僅分析最近幾天的案例
Returns:
帶有聚類標籤的 DataFrame 和聚類標籤陣列
"""
# 篩選最近的案例
cutoff_date = datetime.now() - timedelta(days=recent_days)
df_recent = df[df['case_date'] >= cutoff_date].copy()
print(f"分析最近 {recent_days} 天的案例數: {len(df_recent)}")
# 準備座標資料
coords = df_recent[['latitude', 'longitude']].values
# 計算每度約對應的公里數 (台北地區)
# 緯度: 1度 ≈ 111 公里
# 經度: 1度 ≈ 111 * cos(緯度) 公里
lat_mean = coords[:, 0].mean()
km_per_lat = 111.0
km_per_lon = 111.0 * np.cos(np.radians(lat_mean))
# 將座標轉換為公里單位
coords_km = np.array([
coords[:, 0] * km_per_lat,
coords[:, 1] * km_per_lon
]).T
# 標準化座標
coords_scaled = self.scaler.fit_transform(coords_km)
# 計算對應的 eps (在標準化空間中)
eps_scaled = self.eps_km / np.sqrt((km_per_lat**2 + km_per_lon**2) / 2)
# 執行 DBSCAN 聚類
dbscan = DBSCAN(eps=eps_scaled, min_samples=self.min_samples)
clusters = dbscan.fit_predict(coords_scaled)
# 將聚類標籤加入 DataFrame
df_recent['cluster'] = clusters
# 統計聚類資訊
n_clusters = len(set(clusters)) - (1 if -1 in clusters else 0)
n_noise = list(clusters).count(-1)
print(f"\n識別出的熱點數量: {n_clusters}")
print(f"孤立案例數量: {n_noise}")
# 計算每個聚類的統計資訊
if n_clusters > 0:
print("\n熱點詳細資訊:")
print("-" * 60)
for cluster_id in range(n_clusters):
cluster_cases = df_recent[df_recent['cluster'] == cluster_id]
center_lat = cluster_cases['latitude'].mean()
center_lon = cluster_cases['longitude'].mean()
n_cases = len(cluster_cases)
print(f"熱點 {cluster_id + 1}:")
print(f" 案例數: {n_cases}")
print(f" 中心座標: ({center_lat:.4f}, {center_lon:.4f})")
print(f" 主要行政區: {cluster_cases['district'].mode()[0]}")
print()
return df_recent, clusters
def visualize_hotspots(
self,
df: pd.DataFrame,
output_file: str = "disease_hotspots_map.html"
):
"""
視覺化疾病熱點地圖
Args:
df: 包含聚類標籤的案例資料
output_file: 輸出 HTML 檔案路徑
"""
# 建立台北市地圖
taipei_center = [25.0330, 121.5654]
m = folium.Map(
location=taipei_center,
zoom_start=12,
tiles='OpenStreetMap'
)
# 定義聚類顏色
colors = ['red', 'blue', 'green', 'purple', 'orange',
'darkred', 'darkblue', 'darkgreen', 'cadetblue', 'pink']
# 繪製案例點
for idx, row in df.iterrows():
cluster_id = row['cluster']
if cluster_id == -1:
# 孤立案例用灰色標記
color = 'gray'
popup_text = f"孤立案例 {row['case_id']}"
else:
# 熱點案例用對應顏色標記
color = colors[cluster_id % len(colors)]
popup_text = f"熱點 {cluster_id + 1} - 案例 {row['case_id']}"
folium.CircleMarker(
location=[row['latitude'], row['longitude']],
radius=5,
popup=popup_text,
color=color,
fill=True,
fillColor=color,
fillOpacity=0.7
).add_to(m)
# 標註熱點中心
n_clusters = df['cluster'].max() + 1
for cluster_id in range(n_clusters):
cluster_cases = df[df['cluster'] == cluster_id]
if len(cluster_cases) > 0:
center_lat = cluster_cases['latitude'].mean()
center_lon = cluster_cases['longitude'].mean()
n_cases = len(cluster_cases)
folium.Marker(
location=[center_lat, center_lon],
popup=f"熱點 {cluster_id + 1}<br>案例數: {n_cases}",
icon=folium.Icon(color=colors[cluster_id % len(colors)], icon='info-sign')
).add_to(m)
# 儲存地圖
m.save(output_file)
print(f"\n✅ 熱點地圖已儲存至: {output_file}")
def plot_temporal_trend(
self,
df: pd.DataFrame,
cluster_id: int = None
):
"""
繪製時間趨勢圖
Args:
df: 案例資料
cluster_id: 特定聚類 ID (None 表示所有案例)
"""
if cluster_id is not None:
df_plot = df[df['cluster'] == cluster_id].copy()
title = f"熱點 {cluster_id + 1} 的時間趨勢"
else:
df_plot = df.copy()
title = "整體案例時間趨勢"
# 按日期統計案例數
df_plot['date'] = df_plot['case_date'].dt.date
daily_cases = df_plot.groupby('date').size().reset_index(name='cases')
daily_cases = daily_cases.sort_values('date')
# 繪製趨勢圖
plt.figure(figsize=(12, 6))
plt.plot(daily_cases['date'], daily_cases['cases'], marker='o', linewidth=2)
plt.xlabel('日期', fontsize=12)
plt.ylabel('案例數', fontsize=12)
plt.title(title, fontsize=14, pad=20)
plt.grid(True, alpha=0.3)
plt.xticks(rotation=45)
plt.tight_layout()
filename = f"temporal_trend_cluster_{cluster_id if cluster_id is not None else 'all'}.png"
plt.savefig(filename, dpi=300, bbox_inches='tight')
print(f"時間趨勢圖已儲存至: {filename}")
# 使用範例
if __name__ == "__main__":
print("=" * 60)
print("疾病熱點分析系統")
print("=" * 60)
# 建立分析系統實例
analyzer = DiseaseHotspotAnalyzer(eps_km=1.0, min_samples=5)
# 生成模擬案例資料
print("\n正在生成模擬疾病案例資料...")
df_cases = analyzer.generate_disease_cases(n_cases=500, n_hotspots=3)
print(f"總案例數: {len(df_cases)}")
print(f"時間範圍: {df_cases['case_date'].min()} 至 {df_cases['case_date'].max()}")
# 識別熱點
print("\n" + "=" * 60)
print("執行熱點識別分析")
print("=" * 60)
df_clustered, clusters = analyzer.identify_hotspots(df_cases, recent_days=14)
# 視覺化熱點地圖
analyzer.visualize_hotspots(df_clustered)
# 繪製整體時間趨勢
print("\n" + "=" * 60)
print("繪製時間趨勢圖")
print("=" * 60)
analyzer.plot_temporal_trend(df_clustered)
# 繪製各熱點的時間趨勢
n_clusters = df_clustered['cluster'].max() + 1
for cluster_id in range(n_clusters):
analyzer.plot_temporal_trend(df_clustered, cluster_id=cluster_id)
結語
資料科學正在深刻改變公共衛生的面貌。從學術界的理論創新到產業界的實務應用,資料驅動決策已經證明其能夠創造巨大價值。台灣擁有優秀的資訊技術人才、完善的健保資料基礎設施、高品質的醫療服務體系,在發展資料驅動公共衛生方面具備良好條件。
然而,技術的應用必須伴隨著倫理的考量。資料隱私、演算法公平性、預測準確性的局限,都是我們在推動資料科學應用時必須審慎面對的挑戰。建立完善的資料治理機制、培養跨領域專業人才、促進公私部門的協作,是實現資料驅動公共衛生願景的關鍵要素。
從 Netflix 的內容策略到洛杉磯的預測性警務,跨產業的成功案例提供了寶貴的學習經驗。公共衛生領域可以借鑑這些實踐,同時針對健康議題的特殊性進行調適。未來,隨著人工智慧技術的持續進步、資料整合能力的提升、運算成本的下降,資料科學將在疫情預測、精準介入、資源優化等方面發揮更大作用。
玄貓認為,資料驅動的公共衛生不僅是技術問題,更是組織文化與思維模式的轉變。唯有當資料成為決策的常態依據,當實證精神深植於公衛體系,我們才能真正實現以數據守護全民健康的願景。台灣公衛界應該積極擁抱這個變革浪潮,讓資料科學成為提升國民健康、縮小健康不平等的有力工具。