| name | db-health-check |
| description | 深度資料庫健康檢查，驗證資料品質、新鮮度、一致性。當使用者想檢查系統狀態、 資料是否正常、有無異常時使用。觸發詞: 資料健康, db health, 資料品質, data quality, health check, 資料檢查, 資料庫狀態, 系統狀態, 資料診斷
|
DB Health Check
對 data/data.db 執行全面健康檢查，涵蓋 row counts、新鮮度、NULL rate、orphan records、重複偵測、表間一致性。
執行完整健康檢查
cd "D:/VScode_project/NLP data for trading" && python -c "
import sqlite3
from datetime import datetime, timedelta
conn = sqlite3.connect('data/data.db')
conn.row_factory = sqlite3.Row
c = conn.cursor()
score = 100
issues = []
today = datetime.now().strftime('%Y-%m-%d')
print('=' * 80)
print(' PAM 資料庫健康檢查報告')
print(f' 檢查時間: {datetime.now().strftime(\"%Y-%m-%d %H:%M:%S\")}')
print('=' * 80)
# ============================================================
# 1. 所有表 Row Count + 最新時間戳
# ============================================================
print('\n## 1. 表格統計 (Row Count + 最新紀錄)\n')
tables_ts = {
'congress_trades': 'created_at',
'extraction_log': 'created_at',
'sec_form4_trades': 'created_at',
'signal_quality_scores': 'scored_at',
'convergence_signals': 'detected_at',
'politician_rankings': 'updated_at',
'alpha_signals': 'created_at',
'enhanced_signals': 'updated_at',
'sector_rotation_signals': 'created_at',
'portfolio_positions': 'created_at',
'rebalance_history': 'created_at',
'signal_performance': 'evaluated_at',
'fama_french_results': 'created_at',
'social_posts': 'created_at',
'social_signals': 'created_at',
'ai_intelligence_signals': 'timestamp',
'senate_trades': 'created_at',
'house_trades': 'created_at',
'risk_assessments': 'assessed_at',
}
print(f' {\"Table\":30s} | {\"Rows\":>8s} | {\"Latest\":20s} | Status')
print(' ' + '-' * 80)
for table, ts_col in tables_ts.items():
try:
c.execute(f'SELECT COUNT(*) FROM [{table}]')
count = c.fetchone()[0]
try:
c.execute(f'SELECT MAX([{ts_col}]) FROM [{table}]')
latest = c.fetchone()[0] or 'N/A'
except:
latest = 'N/A'
status = '[OK]'
if count == 0:
status = '[EMPTY]'
issues.append(f'{table}: 表為空')
score -= 2
print(f' {table:30s} | {count:8d} | {str(latest):20s} | {status}')
except Exception as e:
print(f' {table:30s} | {\"MISSING\":>8s} | {\"N/A\":20s} | [MISSING]')
issues.append(f'{table}: 表不存在')
score -= 3
# ============================================================
# 2. 資料新鮮度 (congress_trades)
# ============================================================
print('\n## 2. 資料新鮮度\n')
try:
c.execute('SELECT MAX(transaction_date) as max_td, MAX(filing_date) as max_fd, MAX(created_at) as max_ca FROM congress_trades')
r = c.fetchone()
max_td = r['max_td'] or 'N/A'
max_fd = r['max_fd'] or 'N/A'
max_ca = r['max_ca'] or 'N/A'
print(f' 最新 transaction_date: {max_td}')
print(f' 最新 filing_date: {max_fd}')
print(f' 最新 created_at: {max_ca}')
if max_td != 'N/A':
try:
td_date = datetime.strptime(max_td[:10], '%Y-%m-%d')
days_old = (datetime.now() - td_date).days
print(f' 資料延遲天數: {days_old} 天')
if days_old > 14:
issues.append(f'congress_trades 資料已 {days_old} 天未更新')
score -= 10
print(' [WARN] 資料超過 14 天未更新!')
elif days_old > 7:
issues.append(f'congress_trades 資料已 {days_old} 天未更新')
score -= 5
print(' [WARN] 資料超過 7 天未更新')
else:
print(' [OK] 資料新鮮度正常')
except ValueError:
print(f' [WARN] 無法解析日期格式: {max_td}')
except Exception as e:
print(f' [ERROR] 查詢失敗: {e}')
score -= 10
# ============================================================
# 3. 重要欄位 NULL Rate
# ============================================================
print('\n## 3. 重要欄位 NULL Rate\n')
null_checks = [
('congress_trades', ['politician_name', 'ticker', 'transaction_type', 'transaction_date', 'data_hash']),
('alpha_signals', ['ticker', 'politician_name', 'direction', 'signal_strength', 'confidence']),
('signal_quality_scores', ['politician_name', 'ticker', 'sqs', 'grade']),
('enhanced_signals', ['ticker', 'politician_name', 'enhanced_strength', 'pacs_score']),
]
for table, columns in null_checks:
try:
c.execute(f'SELECT COUNT(*) FROM [{table}]')
total = c.fetchone()[0]
if total == 0:
continue
print(f' [{table}] (total: {total})')
for col in columns:
try:
c.execute(f'SELECT COUNT(*) FROM [{table}] WHERE [{col}] IS NULL')
null_count = c.fetchone()[0]
rate = null_count / total * 100
status = '[OK]' if rate < 5 else '[WARN]' if rate < 20 else '[FAIL]'
if rate >= 5:
issues.append(f'{table}.{col} NULL rate: {rate:.1f}%')
score -= min(5, int(rate / 10))
print(f' {col:25s}: {null_count:5d} NULL ({rate:5.1f}%) {status}')
except:
print(f' {col:25s}: [COLUMN MISSING]')
print()
except:
pass
# ============================================================
# 4. Orphan Records 檢查
# ============================================================
print('\n## 4. Orphan Records (孤立紀錄)\n')
# alpha_signals.trade_id -> congress_trades.id
try:
c.execute('''
SELECT COUNT(*) FROM alpha_signals
WHERE trade_id IS NOT NULL
AND trade_id NOT IN (SELECT id FROM congress_trades)
''')
orphan_alpha = c.fetchone()[0]
status = '[OK]' if orphan_alpha == 0 else '[WARN]'
if orphan_alpha > 0:
issues.append(f'alpha_signals 有 {orphan_alpha} 筆 orphan records (trade_id 無對應)')
score -= 5
print(f' alpha_signals -> congress_trades: {orphan_alpha} orphans {status}')
except Exception as e:
print(f' alpha_signals -> congress_trades: [ERROR] {e}')
# signal_quality_scores.trade_id -> congress_trades.id
try:
c.execute('''
SELECT COUNT(*) FROM signal_quality_scores
WHERE trade_id IS NOT NULL
AND trade_id NOT IN (SELECT id FROM congress_trades)
''')
orphan_sqs = c.fetchone()[0]
status = '[OK]' if orphan_sqs == 0 else '[WARN]'
if orphan_sqs > 0:
issues.append(f'signal_quality_scores 有 {orphan_sqs} 筆 orphan records')
score -= 5
print(f' signal_quality_scores -> congress_trades: {orphan_sqs} orphans {status}')
except Exception as e:
print(f' signal_quality_scores -> congress_trades: [ERROR] {e}')
# signal_performance.signal_id -> alpha_signals.id
try:
c.execute('''
SELECT COUNT(*) FROM signal_performance
WHERE signal_id IS NOT NULL
AND signal_id NOT IN (SELECT id FROM alpha_signals)
''')
orphan_sp = c.fetchone()[0]
status = '[OK]' if orphan_sp == 0 else '[WARN]'
if orphan_sp > 0:
issues.append(f'signal_performance 有 {orphan_sp} 筆 orphan records')
score -= 3
print(f' signal_performance -> alpha_signals: {orphan_sp} orphans {status}')
except Exception as e:
print(f' signal_performance -> alpha_signals: [ERROR] {e}')
# ============================================================
# 5. 重複偵測 (data_hash)
# ============================================================
print('\n## 5. 重複偵測 (data_hash uniqueness)\n')
for table in ['congress_trades', 'sec_form4_trades', 'social_posts']:
try:
c.execute(f'SELECT COUNT(*) FROM [{table}]')
total = c.fetchone()[0]
if total == 0:
print(f' {table}: [EMPTY]')
continue
c.execute(f'SELECT data_hash, COUNT(*) as cnt FROM [{table}] GROUP BY data_hash HAVING cnt > 1')
dupes = c.fetchall()
dupe_count = len(dupes)
dupe_rows = sum(r['cnt'] - 1 for r in dupes) if dupes else 0
status = '[OK]' if dupe_count == 0 else '[WARN]'
if dupe_count > 0:
issues.append(f'{table} 有 {dupe_count} 組重複 hash ({dupe_rows} 多餘筆)')
score -= 5
print(f' {table:25s}: {dupe_count} duplicate groups ({dupe_rows} extra rows) {status}')
except Exception as e:
print(f' {table:25s}: [ERROR] {e}')
# ============================================================
# 6. 表間一致性
# ============================================================
print('\n## 6. 表間一致性\n')
try:
c.execute('SELECT COUNT(*) FROM congress_trades')
ct_count = c.fetchone()[0]
c.execute('SELECT COUNT(*) FROM signal_quality_scores')
sqs_count = c.fetchone()[0]
c.execute('SELECT COUNT(*) FROM alpha_signals')
as_count = c.fetchone()[0]
coverage_sqs = (sqs_count / ct_count * 100) if ct_count > 0 else 0
coverage_alpha = (as_count / ct_count * 100) if ct_count > 0 else 0
print(f' congress_trades: {ct_count:6d} rows')
print(f' signal_quality_scores: {sqs_count:6d} rows (coverage: {coverage_sqs:.1f}%)')
print(f' alpha_signals: {as_count:6d} rows (coverage: {coverage_alpha:.1f}%)')
if coverage_sqs < 80 and ct_count > 0:
issues.append(f'SQS coverage only {coverage_sqs:.1f}% of trades')
score -= 3
# Enhanced vs Alpha
c.execute('SELECT COUNT(*) FROM enhanced_signals')
es_count = c.fetchone()[0]
if as_count > 0:
enh_coverage = es_count / as_count * 100
print(f' enhanced_signals: {es_count:6d} rows (coverage: {enh_coverage:.1f}% of alpha)')
else:
print(f' enhanced_signals: {es_count:6d} rows')
# Portfolio vs Alpha
c.execute('SELECT COUNT(DISTINCT ticker) FROM alpha_signals')
alpha_tickers = c.fetchone()[0]
c.execute('SELECT COUNT(*) FROM portfolio_positions')
port_count = c.fetchone()[0]
print(f' portfolio_positions: {port_count:6d} rows (from {alpha_tickers} unique alpha tickers)')
except Exception as e:
print(f' [ERROR] 一致性檢查失敗: {e}')
# ============================================================
# 7. 日期合理性
# ============================================================
print('\n## 7. 日期合理性\n')
try:
c.execute(\"SELECT COUNT(*) FROM congress_trades WHERE transaction_date > date('now', '+1 day')\")
future = c.fetchone()[0]
c.execute(\"SELECT COUNT(*) FROM congress_trades WHERE transaction_date < '2020-01-01'\")
old = c.fetchone()[0]
status_f = '[OK]' if future == 0 else '[WARN]'
status_o = '[OK]' if old == 0 else '[WARN]'
if future > 0:
issues.append(f'congress_trades 有 {future} 筆未來日期')
score -= 5
if old > 0:
issues.append(f'congress_trades 有 {old} 筆 2020 年以前的日期')
score -= 2
print(f' 未來日期 (transaction_date > today): {future} {status_f}')
print(f' 過舊日期 (transaction_date < 2020): {old} {status_o}')
except Exception as e:
print(f' [ERROR] {e}')
# ============================================================
# 8. Chamber 分布
# ============================================================
print('\n## 8. Chamber 分布\n')
try:
c.execute('SELECT chamber, COUNT(*) as cnt FROM congress_trades GROUP BY chamber')
for r in c.fetchall():
print(f' {(r[\"chamber\"] or \"NULL\"):10s}: {r[\"cnt\"]} trades')
except:
pass
# ============================================================
# SUMMARY
# ============================================================
score = max(0, score)
if score >= 90:
grade = 'A (Excellent)'
elif score >= 75:
grade = 'B (Good)'
elif score >= 60:
grade = 'C (Fair)'
elif score >= 40:
grade = 'D (Poor)'
else:
grade = 'F (Critical)'
print('\n' + '=' * 80)
print(f' 健康評分: {score}/100 ({grade})')
print('=' * 80)
if issues:
print('\n 問題清單:')
for i, issue in enumerate(issues, 1):
print(f' {i}. {issue}')
else:
print('\n [OK] 未發現問題!')
print()
conn.close()
"
特定檢查
根據使用者需求執行特定檢查:
- 只看 row counts: 簡化為 Step 1 查詢
- 特定表: 修改查詢只看指定表
- 修復建議: 針對發現的問題提供 SQL 修復指令
- 歷史趨勢: 比較
extraction_log 中的歷次 ETL 執行結果
常見修復指令
DELETE FROM congress_trades WHERE rowid NOT IN (
SELECT MIN(rowid) FROM congress_trades GROUP BY data_hash
);
DELETE FROM alpha_signals WHERE trade_id NOT IN (
SELECT id FROM congress_trades
);
Output Format
以繁體中文輸出，包含:
- 8 項檢查結果 (附 [OK]/[WARN]/[FAIL] 標記)
- 健康評分 (0-100) 和等級 (A-F)
- 問題清單 + 修復建議
