quantitative trading（量化取引）の世界へようこそ！バックテスト（回測）は、自分の取引戦略が過去のデータで約款に通用するかどうかを検証する最も重要な工程です。しかし、「データ完整性（データ完全性）」という言葉を聞いたことがありますか？データが欠落していたり不正確だったりすると、どんなに優れた戦略であっても意味のない結果になってしまいます。
本記事では、HolySheep AIのAPIを使用して、量化バックテストのデータ完整性を分析方法をお届けします。プログラミングが初めての方も 걱정 하지 마세요 — 完全にゼロから説明します！
データ完整性とは？なぜ重要か
データ完整性とは、バックテストに使用するデータが「正確で」「完整で」「一貫性がある」状態を意味します。具体的には：
- 欠損値がない：価格が突然途切れていない
- タイムスタンプの一貫性：データが全て同じタイムゾーン・時間間隔
- 出来高の妥当性：価格と出来高の関係が物理的に可能
- 異常値の検出：市場暴落やエラーによる異常な値がない
私の実際の経験では、データ完整性チェックを怠ったために3ヶ月分のバックテスト結果を無駄にしたことがあります。特に新興市場の株式データでは、分割調整や配当調整が不十分なケースが多く、戦略の有効性を正確に評価できませんでした。
HolySheep APIとは？
HolySheep AIは、最先端のAIモデルを低コストで利用できるAPI基盤です。量化分析において、こんなに素晴らしいポイントがあります：
| 項目 | HolySheep AI | 一般的なAPI | 節約率 |
|---|---|---|---|
| 汇率 | ¥1=$1 | ¥7.3=$1 | 85%節約 |
| DeepSeek V3.2 | $0.42/MTok | $2.5/MTok | 83%節約 |
| Gemini 2.5 Flash | $2.50/MTok | $7.5/MTok | 67%節約 |
| レイテンシ | <50ms | 100-300ms | 3-6倍高速 |
| 決済方法 | WeChat Pay/Alipay対応 | クレジットカードのみ | удобно |
💡 ヒント：新規登録で無料クレジットが付与されます。今すぐ登録して、成本を試算してみましょう！
向いている人・向いていない人
| 向いている人 | 向いていない人 |
|---|---|
| ✅ 量化取引を始めたばかりの方 | ❌ 超高速のリアルタイム取引（HFT）を目指す方 |
| ✅ バックテストの精度を上げたい方 | ❌ 自前で大量GPUサーバーを維持できる企業 |
| ✅ コスト効率の良いAIツールを探している方 | ❌ オープンソースのみに絞りたい方 |
| ✅ 中国・アジア市場のデータ分析をする方 | ❌ 北米ETFのみを取引する方 |
価格とROI
量化トレーダーにとって、コスト管理同样に重要です。HolySheep AIの料金体系を見てみましょう：
| モデル | 2026年価格 | 用途 | コスト効率 |
|---|---|---|---|
| DeepSeek V3.2 | $0.42/MTok | データ清洗・異常値検出 | ⭐⭐⭐⭐⭐ |
| Gemini 2.5 Flash | $2.50/MTok | パターン分析・報告生成 | ⭐⭐⭐⭐ |
| Claude Sonnet 4.5 | $15/MTok | 複雑な戦略評価 | ⭐⭐⭐ |
| GPT-4.1 | $8/MTok | 汎用分析 | ⭐⭐⭐ |
實際的な計算：1万件のバックテストデータポイントを分析する場合、DeepSeek V3.2を使用すれば約$0.0042で完了します。月間100回の分析を行っても、仅仅$0.42程度。これはCoffee一杯分钱以下です！
HolySheepを選ぶ理由
量化分析の分野でHolySheep AIを選んだ私の理由を分享します：
- コスト削減：従来のAPI比85%の節約は、 个人開発者にとって人生を改变するほど
- 亞洲対応：WeChat PayとAlipayに対応しているため、中国の取引所に簡単に接続可能
- 低レイテンシ：<50msの応答速度は、バックテストの反復作業を大きく加速
- 多样的モデル：DeepSeekからClaudeまで、目的に合わせて最適なモデルを選択可能
ステップバイステップ：APIを使ってデータ完整性分析
準備：APIキーの取得
まず、HolySheep AIに登録して、APIキーを取得します。ダッシュボードで「API Keys」を選択し、新しいキーを生成してください。
📸 スクリーンショットヒント：ダッシュボードの右上に있는「Keys」メニューをクリック → 「Create New Key」ボタンを選択 → キーに名前を付けて「Generate」をクリック
ステップ1：環境設定
まず、必要なライブラリをインストールします。Pythonを使った例を示します：
# 必要なライブラリのインストール
pip install requests pandas python-dotenv
または、requirements.txtに追加する場合
requests>=2.28.0
pandas>=1.5.0
python-dotenv>=0.21.0
ステップ2：基本的なAPI接続
まずはHolySheep AIに正しく接続できるか確認しましょう：
import requests
import json
import os
from dotenv import load_dotenv
.envファイルからAPIキーを読み込み（推奨）
load_dotenv()
API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
BASE_URL = "https://api.holysheep.ai/v1"
def test_connection():
"""API接続の動作確認"""
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
# 単純なCompletions APIで接続テスト
payload = {
"model": "deepseek-chat-v3.2",
"messages": [
{"role": "user", "content": "「接続テスト」と一言で返事してください"}
],
"max_tokens": 50
}
try:
response = requests.post(
f"{BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=10
)
if response.status_code == 200:
result = response.json()
print("✅ 接続成功！")
print(f"応答: {result['choices'][0]['message']['content']}")
print(f"使用トークン: {result.get('usage', {}).get('total_tokens', 'N/A')}")
return True
else:
print(f"❌ エラー: {response.status_code}")
print(f"詳細: {response.text}")
return False
except requests.exceptions.Timeout:
print("❌ タイムアウト: サーバーが応答しません")
return False
except Exception as e:
print(f"❌ 予期しないエラー: {str(e)}")
return False
接続テスト実行
if __name__ == "__main__":
test_connection()📸 期待される出力：「✅ 接続成功！」というメッセージと、使用したトークン数が表示されます。
ステップ3：バックテストデータの準備
分析対象のバックテストデータを作成します。CSVファイルから読み込む例：
import pandas as pd
from datetime import datetime
def load_backtest_data(filepath):
"""
バックテストデータを読み込み、データ完整性を初期チェック
"""
df = pd.read_csv(filepath)
print(f"📊 データ読み込み完了")
print(f" - 総行数: {len(df):,}")
print(f" - 列名: {list(df.columns)}")
print(f" - 期間: {df['date'].min()} ~ {df['date'].max()}")
# 基本的な完整性チェック
issues = []
# 1. 欠損値チェック
missing = df.isnull().sum()
if missing.any():
issues.append(f"欠損値あり: {missing[missing > 0].to_dict()}")
# 2. 異常値チェック（価格が0以下）
if (df['close'] <= 0).any():
issues.append("⚠️ 価格が0以下のデータが存在")
# 3. 出来高の妥当性チェック
if (df['volume'] < 0).any():
issues.append("⚠️ 出来高が負の値が存在")
if issues:
print("🔍 検出された問題:")
for issue in issues:
print(f" - {issue}")
else:
print("✅ 基本的な完整性チェック通過")
return df
使用例（CSVファイルがある場合）
df = load_backtest_data("backtest_results.csv")
ステップ4：AIを使った高度な完整性分析
HolySheep AIのDeepSeek V3.2モデルを使って、より高度なデータ完整性分析を行います：
import requests
import json
import pandas as pd
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY" # 実際のキーに置き換えてください
def analyze_data_integrity_with_ai(df, symbol="BTC/USDT"):
"""
HolySheep AIを使ってバックテストデータの完整性を分析
"""
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
# 分析用のデータを準備（最初の100行をサンプル）
sample_df = df.head(100)
data_summary = {
"symbol": symbol,
"total_records": len(df),
"columns": list(df.columns),
"sample_data": sample_df.to_dict(orient='records'),
"basic_stats": {
"price_range": {
"open": {"min": float(df['open'].min()), "max": float(df['open'].max())},
"high": {"min": float(df['high'].min()), "max": float(df['high'].max())},
"low": {"min": float(df['low'].min()), "max": float(df['low'].max())},
"close": {"min": float(df['close'].min()), "max": float(df['close'].max())}
},
"volume_stats": {
"mean": float(df['volume'].mean()),
"std": float(df['volume'].std()),
"min": float(df['volume'].min()),
"max": float(df['volume'].max())
}
}
}
# AIへのプロンプト
prompt = f"""
あなたは量化取引のデータ分析专家です。以下のバックテストデータの完整性を分析してください。
【シンボル】{symbol}
【総レコード数】{len(df):,}
【カラム】{', '.join(data_summary['columns'])}
【価格統計】
- 始値: {data_summary['basic_stats']['price_range']['open']}
- 高値: {data_summary['basic_stats']['price_range']['high']}
- 安値: {data_summary['basic_stats']['price_range']['low']}
- 終値: {data_summary['basic_stats']['price_range']['close']}
【出来高統計】
- 平均: {data_summary['basic_stats']['volume_stats']['mean']:.2f}
- 標準偏差: {data_summary['basic_stats']['volume_stats']['std']:.2f}
- 範囲: {data_summary['basic_stats']['volume_stats']['min']:.2f} ~ {data_summary['basic_stats']['volume_stats']['max']:.2f}
【サンプルデータ（先頭5件）】
{json.dumps(sample_df.head().to_dict(orient='records'), indent=2)}
以下の点についてJSONフォーマットで報告してください：
1. data_quality_score (0-100): データ完整性のスコア
2. issues_found: 発見された問題のリスト
3. recommendations: 改善の提案
4. is_ready_for_backtest: バックテストに使用可能か（true/false）
必ず有効なJSONのみを返してください。
"""
payload = {
"model": "deepseek-chat-v3.2",
"messages": [
{"role": "system", "content": "あなたは量化取引のデータ分析专家です。必ず有効なJSONのみを返してください。"},
{"role": "user", "content": prompt}
],
"temperature": 0.3, # 結果の一貫性のため低めに設定
"max_tokens": 1500
}
try:
print("🔄 AI分析中...")
response = requests.post(
f"{BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
if response.status_code == 200:
result = response.json()
analysis = result['choices'][0]['message']['content']
usage = result.get('usage', {})
print("\n" + "="*50)
print("📋 AI分析結果")
print("="*50)
# JSONをパースして表示
try:
analysis_json = json.loads(analysis)
print(f"\n🎯 データ品質スコア: {analysis_json.get('data_quality_score', 'N/A')}/100")
print(f"\n✅ バックテスト利用可能: {'はい' if analysis_json.get('is_ready_for_backtest') else 'いいえ'}")
if analysis_json.get('issues_found'):
print("\n🔍 検出された問題:")
for i, issue in enumerate(analysis_json['issues_found'], 1):
print(f" {i}. {issue}")
if analysis_json.get('recommendations'):
print("\n💡 改善提案:")
for i, rec in enumerate(analysis_json['recommendations'], 1):
print(f" {i}. {rec}")
except json.JSONDecodeError:
print(analysis) # JSON以外で返ってきた場合はそのまま表示
print(f"\n💰 使用トークン: {usage.get('total_tokens', 'N/A')}")
return analysis
else:
print(f"❌ APIエラー: {response.status_code}")
print(f"詳細: {response.text}")
return None
except Exception as e:
print(f"❌ エラー: {str(e)}")
return None
使用例
if __name__ == "__main__":
# サンプルデータ作成（実際にはCSVから読み込む）
import numpy as np
from datetime import datetime, timedelta
dates = [(datetime(2024, 1, 1) + timedelta(days=i)) for i in range(500)]
sample_data = pd.DataFrame({
'date': dates,
'open': np.random.uniform(40000, 50000, 500),
'high': np.random.uniform(42000, 52000, 500),
'low': np.random.uniform(38000, 48000, 500),
'close': np.random.uniform(40000, 50000, 500),
'volume': np.random.uniform(1000, 10000, 500)
})
# AI分析を実行
analyze_data_integrity_with_ai(sample_data, symbol="BTC/USDT")📸 期待される出力：データ品質スコア、検出された問題一覧、改善提案が詳しく表示されます。
Tardis 量化プラットフォームとの連携
Tardis（ターディズ）は、加密通貨のデータ配信で知られるプラットフォームです。TardisからエクスポートしたデータをHolySheep AIで完整性分析するワークフローを紹介します：
import requests
import pandas as pd
from datetime import datetime
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
def process_tardis_export(csv_path, api_key=API_KEY):
"""
TardisからエクスポートしたCSVをHolySheep AIで分析
Args:
csv_path: TardisからエクスポートしたCSVファイルのパス
api_key: HolySheep APIキー
"""
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
# 1. Tardisデータの読み込み
print("📥 Tardisデータを読み込み中...")
df = pd.read_csv(csv_path)
# Tardis形式の必要カラムを確認
required_columns = ['timestamp', 'open', 'high', 'low', 'close', 'volume']
missing_cols = [col for col in required_columns if col not in df.columns]
if missing_cols:
print(f"⚠️ 不足しているカラム: {missing_cols}")
# カラム名のマッピングを試行
df = df.rename(columns={
'time': 'timestamp',
'Time': 'timestamp',
'datetime': 'timestamp'
})
print(f"✅ {len(df):,}件のデータを読み込み完了")
# 2. データ完整性の基本的な検証
integrity_report = {
"total_records": len(df),
"missing_values": df.isnull().sum().to_dict(),
"price_anomalies": {
"zero_prices": int((df['close'] == 0).sum()),
"negative_prices": int((df['close'] < 0).sum()),
"high_below_low": int((df['high'] < df['low']).sum())
},
"volume_anomalies": {
"negative_volume": int((df['volume'] < 0).sum()),
"zero_volume": int((df['volume'] == 0).sum())
},
"timestamp_gaps": detect_timestamp_gaps(df['timestamp'].tolist())
}
print("\n📊 初步完整性レポート:")
print(f" - 総レコード: {integrity_report['total_records']:,}")
print(f" - 欠損値: {sum(integrity_report['missing_values'].values())}")
print(f" - 価格異常: {sum(integrity_report['price_anomalies'].values())}")
print(f" - 出来高異常: {sum(integrity_report['volume_anomalies'].values())}")
# 3. AIによる詳細分析
prompt = f"""
Tardisプラットフォームからエクスポートされた暗号通貨バックテストデータを分析してください。
【データ概要】
- レコード数: {len(df):,}
- 期間: {df['timestamp'].iloc[0]} ~ {df['timestamp'].iloc[-1]}
- 欠損値: {integrity_report['missing_values']}
- 価格異常: {integrity_report['price_anomalies']}
- 出来高異常: {integrity_report['volume_anomalies']}
- タイムスタンプギャップ: {integrity_report['timestamp_gaps']}
【価格範囲】
- 高値最高: {df['high'].max():.2f}
- 安値最低: {df['low'].min():.2f}
- 平均出来高: {df['volume'].mean():.2f}
以下のJSONを返してください：
{{
"integrity_score": 0-100のスコア,
"critical_issues": ["致命的な問題のリスト"],
"warnings": ["警告のリスト"],
"data_cleaning_steps": ["推奨される清洗手順"],
"backtest_validity": "このデータのバックテスト有効性について"
}}
"""
payload = {
"model": "deepseek-chat-v3.2",
"messages": [
{"role": "system", "content": "あなたは量化取引とデータ分析の专家です。必ず有効なJSONのみを返してください。"},
{"role": "user", "content": prompt}
],
"temperature": 0.2,
"max_tokens": 1200
}
print("\n🤖 AI詳細分析を実行中...")
response = requests.post(
f"{BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
if response.status_code == 200:
result = response.json()
analysis = result['choices'][0]['message']['content']
usage = result.get('usage', {})
print("\n" + "="*60)
print("🔍 Tardisデータ AI分析結果")
print("="*60)
try:
analysis_json = json.loads(analysis)
print(f"\n🎯 完整性スコア: {analysis_json.get('integrity_score', 'N/A')}/100")
if analysis_json.get('critical_issues'):
print("\n🔴 致命的問題:")
for issue in analysis_json['critical_issues']:
print(f" ⚠️ {issue}")
if analysis_json.get('warnings'):
print("\n🟡 警告:")
for warning in analysis_json['warnings']:
print(f" ⚡ {warning}")
if analysis_json.get('data_cleaning_steps'):
print("\n🧹 推奨清洗手順:")
for step in analysis_json['data_cleaning_steps']:
print(f" {step}")
print(f"\n📝 バックテスト有効性: {analysis_json.get('backtest_validity', 'N/A')}")
except json.JSONDecodeError:
print(analysis)
print(f"\n💰 コスト: ${usage.get('total_tokens', 0) * 0.00042:.4f} (DeepSeek V3.2)")
return integrity_report
def detect_timestamp_gaps(timestamps):
"""タイムスタンプのギャップを検出"""
gaps = []
for i in range(1, len(timestamps)):
try:
t1 = pd.to_datetime(timestamps[i-1])
t2 = pd.to_datetime(timestamps[i])
diff = (t2 - t1).total_seconds()
# 1時間以上のギャップを検出
if diff > 3600:
gaps.append({
"from": str(timestamps[i-1]),
"to": str(timestamps[i]),
"gap_hours": diff / 3600
})
except:
continue
return {"count": len(gaps), "samples": gaps[:5]} # 最初の5件を返す
使用例
report = process_tardis_export("tardis_export.csv")
よくあるエラーと対処法
エラー1：APIキーが無効です（401 Unauthorized）
# ❌ エラー例
{"error": {"message": "Invalid API key provided", "type": "invalid_request_error"}}
✅ 解決策
1. APIキーが正しくコピーされているか確認
2. .envファイルのパスが正しいか確認
3. キーが有効期限切れでないか確認
import os
from dotenv import load_dotenv
.envファイルの内容確認用
load_dotenv()
api_key = os.getenv("HOLYSHEEP_API_KEY")
if not api_key:
print("❌ HOLYSHEEP_API_KEYが.envファイルに設定されていません")
print(" .envファイルを作成して以下を追加してください:")
print(" HOLYSHEEP_API_KEY=あなたの実際のAPIキー")
elif api_key == "YOUR_HOLYSHEEP_API_KEY":
print("❌ プレースホルダーのままです。実際のAPIキーに置き換えてください")
else:
print("✅ APIキーが設定されています")エラー2：タイムアウトエラー（408 Request Timeout）
# ❌ エラー例
requests.exceptions.ReadTimeout: HTTPSConnectionPool
(host='api.holysheep.ai') - Read timed out
✅ 解決策：タイムアウト時間を延長
import requests
方法1：デフォルトのタイムアウトを設定
response = requests.post(
f"{BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=60 # 60秒に延長
)
方法2：接続タイムアウトと読み取りタイムアウトを個別に設定
response = requests.post(
f"{BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=(10, 120) # (接続タイムアウト, 読み取りタイムアウト)
)
方法3：再試行ロジックを実装
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def create_session_with_retry():
session = requests.Session()
retry = Retry(
total=3,
backoff_factor=1,
status_forcelist=[500, 502, 503, 504]
)
adapter = HTTPAdapter(max_retries=retry)
session.mount('https://', adapter)
return session
使用
session = create_session_with_retry()
response = session.post(f"{BASE_URL}/chat/completions", headers=headers, json=payload)エラー3：JSONパースエラー（Response Parsing Error）
# ❌ エラー例
json.JSONDecodeError: Expecting value: line 1 column 1 (char 0)
✅ 解決策：レスポンスの 검증とフォールバック
import json
def safe_json_parse(response_text):
"""JSONパースを安全に行い、失敗時はフォールバック"""
try:
return json.loads(response_text)
except json.JSONDecodeError as e:
print(f"⚠️ JSONパースエラー: {e}")
print(f" レスポンス内容: {response_text[:200]}...")
# マークダウンブロック内のJSONを抽出を試行
import re
json_match = re.search(r'``(?:json)?\s*(\{.*?\})\s*``', response_text, re.DOTALL)
if json_match:
try:
return json.loads(json_match.group(1))
except:
pass
# 最後の braces からJSONを抽出
json_match = re.search(r'\{[\s\S]*\}', response_text)
if json_match:
try:
return json.loads(json_match.group(0))
except:
pass
return None
使用例
response = requests.post(f"{BASE_URL}/chat/completions", headers=headers, json=payload)
result = safe_json_parse(response.text)
if result:
print("✅ JSON解析成功")
else:
print("❌ JSON解析失敗 - レスポンスを而生で処理")
# 生テキストを表示
print(response.text)エラー4：データ型エラー（Type Error）
# ❌ エラー例
TypeError: Object of type float64 is not JSON serializable
✅ 解決策：NumPy/Pandas型をPython標準型に変換
import numpy as np
import pandas as pd
def convert_to_json_serializable(obj):
"""NumPy/Pandas型をJSONシリアライズ可能な型に変換"""
if isinstance(obj, pd.DataFrame):
return obj.to_dict(orient='records')
elif isinstance(obj, pd.Series):
return obj.to_dict()
elif isinstance(obj, np.integer):
return int(obj)
elif isinstance(obj, np.floating):
return float(obj)
elif isinstance(obj, np.ndarray):
return obj.tolist()
elif isinstance(obj, np.bool_):
return bool(obj)
elif isinstance(obj, dict):
return {k: convert_to_json_serializable(v) for k, v in obj.items()}
elif isinstance(obj, list):
return [convert_to_json_serializable(i) for i in obj]
else:
return obj
使用例
data = {
"mean_price": df['close'].mean(), # numpy.float64
"record_count": len(df), # numpy.int64
"is_valid": True # numpy.bool_
}
json_data = convert_to_json_serializable(data)
print(json.dumps(json_data, indent=2)) # ✅ 正しく動作エラー5：レートリミットエラー（429 Too Many Requests）
# ❌ エラー例
{"error": {"message": "Rate limit exceeded", "type": "rate_limit_error"}}
✅ 解決策：指数バックオフで再試行
import time
import random
def request_with_retry(func, max_retries=5, base_delay=1):
"""指数バックオフでリクエストをリトライ"""
for attempt in range(max_retries):
try:
response = func()
if response.status_code == 200:
return response
elif response.status_code == 429:
# レートリミットのレスポンスヘッダーを確認
retry_after = int(response.headers.get('Retry-After', 60))
wait_time = retry_after + random.uniform(0, 5)
print(f"⏳ レートリミット到達。{wait_time:.1f}秒後に再試行...")
time.sleep(wait_time)
else:
print(f"❌ エラー: {response.status_code}")
return response
except Exception as e:
if attempt < max_retries - 1:
wait_time = base_delay * (2 ** attempt) + random.uniform(0, 1)
print(f"⏳ エラー: {e}. {wait_time:.1f}秒後に再試行...")
time.sleep(wait_time)
else:
raise
return None
使用例
def api_call():
return requests.post(f"{BASE_URL}/chat/completions", headers=headers, json=payload)
response = request_with_retry(api_call)実践的な分析ワークフロー
実際に私が每周末に行っている分析ワークフローを共有します：
"""
週次バックテストデータ完整性分析ワークフロー
HolySheep AI × Tardis 連携
"""
import requests
import pandas as pd
import json
from datetime import datetime, timedelta
from pathlib import Path
class WeeklyIntegrityAnalyzer:
def __init__(self, api_key):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.analysis_history = []
def run_weekly_analysis(self, data_dir="data/tardis_exports"):
"""週次分析のメイン関数"""
print("="*60)
print(f"📅 週次データ完整性分析: {datetime.now().strftime('%Y-%m-%d')}")
print("="*60)
data_path = Path(data_dir)
csv_files = list(data_path.glob("*.csv"))
if not csv_files:
print(f"⚠️ {data_dir}にCSVファイルが見つかりません")
return None
results = []
total_cost = 0
for csv_file in csv_files:
print(f"\n📄 分析中: {csv_file.name}")
result = self.analyze_single_file(csv_file)
if result:
results.append(result)
total_cost += result.get('cost', 0)
# サマリー作成
summary = self.create_summary(results)
print("\n" + "="*60)
print("📊 週次サマリー")
print("="*60)
print(f"分析ファイル数: {len(results)}")
print(f"総コスト: ${total_cost:.4f}")
print(f"平均完整性スコア: {summary['avg_integrity_score']:.1f}")
print(f"致命的問題数: {summary['critical_issues_count']}")
return summary
def analyze_single_file(self, file_path):
"""单个ファイルの分析"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
# データ読み込み
df = pd.read_csv(file_path)
# 簡易完整性チェック
issues = {
"missing": df.isnull().sum().sum(),
"anomalies": (df['close'] <= 0).sum(),
"gaps": self._check_gaps(df)
}
# AI分析
prompt = self._create_analysis_prompt(df, issues)
payload = {
"model": "deepseek-chat-v3.2",
"messages": [
{"role": "system", "content": "JSONのみを返してください。"},
{"role": "user", "content": prompt}
],
"temperature": 0.2,
"max_tokens": 800
}
try:
start_time = datetime.now()
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
elapsed = (datetime.now() - start_time).total_seconds()
if response.status_code == 200:
result = response.json()
content = result['choices'][0]['message']['content']
usage = result.get('usage', {})
# コスト計算（DeepSeek V3.2: $0.42/MTok）
cost = usage.get('total_tokens', 0) * 0.00042 / 1000
return {
"file": str(file_path