Claude Codeで15種のデータ分析モジュールを1日5種ペースで量産した方法

1
signalsConfig.py        ← モジュールのパラメータ設定
2
    ↓
3
analyticsEngine.py      ← 分析エンジン本体
4
    ↓
5
mgmtTechnicalIndicators.py  ← 全モジュールの一元管理
6
    ↓
7
pipeline/main.py        ← パイプラインへの組み込み
8
    ↓
9
putAnalyticsDatas/main.py   ← 分析結果のDB格納
10
    ↓
11
dev_analyticsdatas.sql  ← DB DDL（カラム追加）

1
## 分析モジュール追加時の連鎖更新ルール
2
1. signalsConfig.py にパラメータ定義を追加
3
2. mgmt{モジュール名}.py を新規作成（mgmtModuleA.pyをテンプレートとして使用）
4
3. moduleManager.py にimportと呼び出しを追加
5
4. engine.py にモジュールの呼び出しを追加
6
5. pipeline/main.py にモジュール結果の参照を追加
7
6. putAnalyticsDatas/main.py にDB格納ロジックを追加
8
7. dev_analyticsdatas.sql にカラム定義を追加
9

10
※ 1つでも漏れるとパイプライン全体が停止する。全ファイルの更新を確認すること。

1
# mgmtModuleA.py の構造（テンプレート）
2

3
def calculate(df, period=14):
4
    """分析モジュールの計算"""
5
    delta = df['value'].diff()
6
    positive = delta.where(delta > 0, 0)
7
    negative = -delta.where(delta < 0, 0)
8
    avg_pos = positive.rolling(window=period).mean()
9
    avg_neg = negative.rolling(window=period).mean()
10
    ratio = avg_pos / avg_neg
11
    score = 100 - (100 / (1 + ratio))
12
    return score
13

14
def generate_result(df, config):
15
    """検出条件を判定する"""
12 collapsed lines
16
    score = calculate(df, config['period'])
17
    # アラート条件: スコアが下限閾値以下
18
    # 検出条件: スコアが上限閾値以上
19
    results = []
20
    for i in range(len(score)):
21
        if score.iloc[i] < config['lower_threshold']:
22
            results.append('ALERT')
23
        elif score.iloc[i] > config['upper_threshold']:
24
            results.append('DETECT')
25
        else:
26
            results.append('NORMAL')
27
    return results

1
mgmtModuleA.pyをテンプレートにして、順位相関分析のモジュールを作成してください。
2
- 計算期間は9, 26, 52の3パターン
3
- 検出条件: 短期値が-80以下でアラート、+80以上で検出
4
- 6ファイル連鎖更新をすべて実施してください

1
☐ 順位相関分析モジュール作成
2
☐ ダブルピーク/ボトム検出モジュール作成
3
☐ トレンド構造分析モジュール作成
4
☐ 収束/平行パターン検出モジュール作成
5
☐ レンジ判定分析モジュール作成

1
1. Claude Codeにテンプレート+パラメータを指示（2分）
2
2. 6ファイルの自動更新を確認（5分）
3
3. ユニットテスト実行（3分）
4
4. パイプライン実行で結果確認（10分）
5
5. 微調整（パラメータ閾値の変更など）（10-20分）
6
---
7
合計: 30-60分/モジュール

1
# 三点ピークの検出ロジック（簡略版）
2
# 「3つの山のうち中央が最も高い」をどの程度の誤差で判定するか
3
def detect_triple_peak(df, tolerance=0.02):
4
    peaks = find_local_peaks(df['value'])
5
    if len(peaks) < 3:
6
        return None
7
    left, center, right = peaks[-3], peaks[-2], peaks[-1]
8
    # 中央の山が左右より高いか
9
    if df['value'].iloc[center] > df['value'].iloc[left] * (1 + tolerance):
10
        if df['value'].iloc[center] > df['value'].iloc[right] * (1 + tolerance):
11
            return {'type': 'triple_peak', 'center': center}
12
    return None