Change to continuous data with line graph visualization

- Generate continuous time-series data with sine wave patterns
- Show actual data points using line graphs
- Add drift point markers with red dashed lines
- Update analyzer to work with continuous data

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

analyzer.py

@@ -11,21 +11,11 @@ def analyze_drift(X: np.ndarray, y: np.ndarray, drift_points: np.ndarray, drift_
         "drift_locations": drift_points.tolist() if len(drift_points) > 0 else [],
     }
 
-    # incremental drift는 연속 값, 나머지는 이진 분류
-    if drift_type == "incremental":
-        # 연속 값 분석
-        analysis["mean_y"] = float(np.mean(y))
-        analysis["std_y"] = float(np.std(y))
-        analysis["min_y"] = float(np.min(y))
-        analysis["max_y"] = float(np.max(y))
-    else:
-        # 이진 분류 분석
-        class_0_count = int(np.sum(y == 0))
-        class_1_count = int(np.sum(y == 1))
-        analysis["class_0_count"] = class_0_count
-        analysis["class_1_count"] = class_1_count
-        analysis["class_0_ratio"] = float(class_0_count / len(y))
-        analysis["class_1_ratio"] = float(class_1_count / len(y))
+    # 전체 통계 (모든 drift type을 연속 값으로 처리)
+    analysis["mean_y"] = float(np.mean(y))
+    analysis["std_y"] = float(np.std(y))
+    analysis["min_y"] = float(np.min(y))
+    analysis["max_y"] = float(np.max(y))
 
     # 세그먼트별 분석
     segments = []
@@ -36,31 +26,26 @@ def analyze_drift(X: np.ndarray, y: np.ndarray, drift_points: np.ndarray, drift_
         end = segment_boundaries[i + 1]
 
         segment_y = y[start:end]
+        segment_X = X[start:end]
 
-        if drift_type == "incremental":
-            # 연속 값 세그먼트 분석
-            segments.append({
-                "segment_id": i,
-                "start_idx": int(start),
-                "end_idx": int(end),
-                "mean": float(np.mean(segment_y)),
-                "std": float(np.std(segment_y))
-            })
+        # 선형 회귀 계수 계산
+        if len(segment_X) > 1:
+            coeffs = np.polyfit(segment_X, segment_y, 1)
+            slope = float(coeffs[0])
+            intercept = float(coeffs[1])
         else:
-            # 이진 분류 세그먼트 분석
-            class_0_count = int(np.sum(segment_y == 0))
-            class_1_count = int(np.sum(segment_y == 1))
-            total = len(segment_y)
-
-            segments.append({
-                "segment_id": i,
-                "start_idx": int(start),
-                "end_idx": int(end),
-                "class_0_count": class_0_count,
-                "class_1_count": class_1_count,
-                "class_0_ratio": float(class_0_count / total) if total > 0 else 0.0,
-                "class_1_ratio": float(class_1_count / total) if total > 0 else 0.0
-            })
+            slope = 0.0
+            intercept = float(segment_y[0]) if len(segment_y) > 0 else 0.0
+
+        segments.append({
+            "segment_id": i,
+            "start_idx": int(start),
+            "end_idx": int(end),
+            "mean": float(np.mean(segment_y)),
+            "std": float(np.std(segment_y)),
+            "slope": slope,
+            "intercept": intercept
+        })
 
     analysis["segments"] = segments
 
@@ -80,32 +65,19 @@ def format_analysis_summary(analysis: Dict) -> str:
 **전체 데이터:**
 - 총 샘플 수: {analysis['total_samples']}
 - 드리프트 발생 횟수: {analysis['num_drift_points']}
-"""
-
-    if drift_type == "incremental":
-        summary += f"""- 평균 값: {analysis['mean_y']:.2f}
+- 평균 값: {analysis['mean_y']:.2f}
 - 표준편차: {analysis['std_y']:.2f}
 - 범위: [{analysis['min_y']:.2f}, {analysis['max_y']:.2f}]
-"""
-    else:
-        summary += f"""- Class 0 (파란색): {analysis['class_0_count']} 샘플 ({analysis['class_0_ratio']*100:.1f}%)
-- Class 1 (초록색): {analysis['class_1_count']} 샘플 ({analysis['class_1_ratio']*100:.1f}%)
-"""
 
-    summary += "\n**세그먼트별 분석:**\n"
+**세그먼트별 분석:**
+"""
 
     for seg in analysis['segments']:
-        if drift_type == "incremental":
-            summary += f"""
+        summary += f"""
 **세그먼트 {seg['segment_id'] + 1}** (샘플 {seg['start_idx']}-{seg['end_idx']})
 - 평균: {seg['mean']:.2f}
 - 표준편차: {seg['std']:.2f}
-"""
-        else:
-            summary += f"""
-**세그먼트 {seg['segment_id'] + 1}** (샘플 {seg['start_idx']}-{seg['end_idx']})
-- Class 0: {seg['class_0_count']} 샘플 ({seg['class_0_ratio']*100:.1f}%)
-- Class 1: {seg['class_1_count']} 샘플 ({seg['class_1_ratio']*100:.1f}%)
+- 트렌드: y = {seg['slope']:.4f}x + {seg['intercept']:.2f}
 """
 
     return summary

drift_simulator.py

@@ -3,14 +3,14 @@ from typing import Tuple
 
 def generate_sudden_drift(n_samples: int = 1000, drift_point: int = 500) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
     """급격한 드리프트: t 시점에서 갑자기 데이터 분포 변경"""
-    X = np.arange(n_samples)  # 시간 인덱스
-    y = np.zeros(n_samples, dtype=int)
+    X = np.arange(n_samples)
+    y = np.zeros(n_samples)
 
-    # Before drift: class 0 (파란색)
-    y[:drift_point] = 0
+    # Before drift: y = 2 + sin(X/50) + noise
+    y[:drift_point] = 2 + np.sin(X[:drift_point] / 50) + np.random.normal(0, 0.3, drift_point)
 
-    # After drift: class 1 (초록색)
-    y[drift_point:] = 1
+    # After drift: y = 5 - sin(X/50) + noise (완전히 다른 패턴)
+    y[drift_point:] = 5 - np.sin(X[drift_point:] / 50) + np.random.normal(0, 0.3, n_samples - drift_point)
 
     drift_points = np.array([drift_point])
     return X, y, drift_points
@@ -18,41 +18,42 @@ def generate_sudden_drift(n_samples: int = 1000, drift_point: int = 500) -> Tupl
 
 def generate_gradual_drift(n_samples: int = 1000, drift_start: int = 300, drift_end: int = 700) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
     """점진적 드리프트: 두 분포가 섞이며 천천히 전환"""
-    X = np.arange(n_samples)  # 시간 인덱스
-    y = np.zeros(n_samples, dtype=int)
+    X = np.arange(n_samples)
+    y = np.zeros(n_samples)
 
-    # Before drift: class 0 (파란색)
-    y[:drift_start] = 0
+    # Before drift: y = 2 + sin(X/50) + noise
+    y[:drift_start] = 2 + np.sin(X[:drift_start] / 50) + np.random.normal(0, 0.3, drift_start)
 
-    # Gradual transition: class 0과 class 1이 섞임
+    # Gradual transition: 점진적으로 변환
     transition_length = drift_end - drift_start
     for i in range(drift_start, drift_end):
-        # 점진적으로 class 1의 비율 증가
         weight = (i - drift_start) / transition_length
-        y[i] = 1 if np.random.random() < weight else 0
+        old_concept = 2 + np.sin(X[i] / 50) + np.random.normal(0, 0.3)
+        new_concept = 5 - np.sin(X[i] / 50) + np.random.normal(0, 0.3)
+        y[i] = (1 - weight) * old_concept + weight * new_concept
 
-    # After drift: class 1 (초록색)
-    y[drift_end:] = 1
+    # After drift: y = 5 - sin(X/50) + noise
+    y[drift_end:] = 5 - np.sin(X[drift_end:] / 50) + np.random.normal(0, 0.3, n_samples - drift_end)
 
     drift_points = np.array([drift_start, drift_end])
     return X, y, drift_points
 
 
-def generate_incremental_drift(n_samples: int = 1000, n_steps: int = 10) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
+def generate_incremental_drift(n_samples: int = 1000, n_steps: int = 5) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
     """증분적 드리프트: 계단식으로 작은 변화가 누적"""
-    X = np.arange(n_samples)  # 시간 인덱스
-    y = np.zeros(n_samples)  # 연속 값 (시각화를 위해)
+    X = np.arange(n_samples)
+    y = np.zeros(n_samples)
 
-    step_size = n_samples // n_steps
+    step_size = n_samples // (n_steps + 1)
     drift_points = []
 
-    for step in range(n_steps):
+    for step in range(n_steps + 1):
         start_idx = step * step_size
-        end_idx = (step + 1) * step_size if step < n_steps - 1 else n_samples
+        end_idx = (step + 1) * step_size if step < n_steps else n_samples
 
-        # 각 단계마다 0에서 1로 점진적 변화
-        value = step / (n_steps - 1)
-        y[start_idx:end_idx] = value
+        # 각 단계마다 평균이 조금씩 변화
+        mean_shift = 2 + (step / n_steps) * 3  # 2에서 5로 점진적 변화
+        y[start_idx:end_idx] = mean_shift + np.sin(X[start_idx:end_idx] / 50) + np.random.normal(0, 0.3, end_idx - start_idx)
 
         if step > 0:
             drift_points.append(start_idx)
@@ -62,8 +63,8 @@ def generate_incremental_drift(n_samples: int = 1000, n_steps: int = 10) -> Tupl
 
 def generate_recurring_drift(n_samples: int = 1000, cycle_length: int = 250) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
     """반복적 드리프트: 이전 분포가 주기적으로 재등장"""
-    X = np.arange(n_samples)  # 시간 인덱스
-    y = np.zeros(n_samples, dtype=int)
+    X = np.arange(n_samples)
+    y = np.zeros(n_samples)
 
     drift_points = []
 
@@ -71,11 +72,11 @@ def generate_recurring_drift(n_samples: int = 1000, cycle_length: int = 250) ->
         cycle_pos = i % cycle_length
 
         if cycle_pos < cycle_length // 2:
-            # Concept A: class 0 (파란색)
-            y[i] = 0
+            # Concept A: y = 2 + sin(X/50) + noise
+            y[i] = 2 + np.sin(X[i] / 50) + np.random.normal(0, 0.3)
         else:
-            # Concept B: class 1 (초록색)
-            y[i] = 1
+            # Concept B: y = 5 - sin(X/50) + noise
+            y[i] = 5 - np.sin(X[i] / 50) + np.random.normal(0, 0.3)
 
         if cycle_pos == cycle_length // 2:
             drift_points.append(i)

visualizer.py

@@ -7,59 +7,30 @@ def create_drift_visualization(X: np.ndarray, y: np.ndarray, drift_points: np.nd
 
     fig = go.Figure()
 
-    # incremental drift는 연속 값으로 처리
-    if drift_type == "incremental":
-        # 색상 리스트 생성
-        colors = []
-        for val in y:
-            # 파란색(0)에서 초록색(1)로 점진적 변화
-            r = int(50 + 50 * val)
-            g = int(100 + 79 * val)
-            b = int(200 - 87 * val)
-            colors.append(f'rgb({r}, {g}, {b})')
-
-        # Scatter로 표현 (색상 영역)
-        fig.add_trace(go.Scatter(
-            x=X,
-            y=np.ones(len(X)),
-            mode='markers',
-            marker=dict(
-                color=colors,
-                size=10,
-                symbol='square',
-                line=dict(width=0)
-            ),
-            showlegend=False,
-            hovertemplate='Time: %{x}<br>Value: %{customdata:.2f}<extra></extra>',
-            customdata=y
-        ))
-    else:
-        # 이진 분류 (0: 파란색, 1: 초록색)
-        # Class 0 (파란색)
-        class_0_mask = y == 0
-        if np.any(class_0_mask):
-            fig.add_trace(go.Scatter(
-                x=X[class_0_mask],
-                y=np.ones(np.sum(class_0_mask)),
-                mode='markers',
-                marker=dict(color='rgb(65, 105, 225)', size=10, symbol='square', line=dict(width=0)),
-                name='Class 0',
-                showlegend=True,
-                hovertemplate='Time: %{x}<br>Class: 0<extra></extra>'
-            ))
-
-        # Class 1 (초록색)
-        class_1_mask = y == 1
-        if np.any(class_1_mask):
-            fig.add_trace(go.Scatter(
-                x=X[class_1_mask],
-                y=np.ones(np.sum(class_1_mask)),
-                mode='markers',
-                marker=dict(color='rgb(50, 205, 50)', size=10, symbol='square', line=dict(width=0)),
-                name='Class 1',
-                showlegend=True,
-                hovertemplate='Time: %{x}<br>Class: 1<extra></extra>'
-            ))
+    # 메인 라인 그래프
+    fig.add_trace(go.Scatter(
+        x=X,
+        y=y,
+        mode='lines+markers',
+        name='Data',
+        line=dict(color='rgb(100, 140, 200)', width=2),
+        marker=dict(size=4, color='rgb(100, 140, 200)'),
+        hovertemplate='Time: %{x}<br>Value: %{y:.2f}<extra></extra>'
+    ))
+
+    # 드리프트 발생 지점 표시
+    y_min, y_max = y.min(), y.max()
+    y_range = y_max - y_min
+
+    for i, drift_point in enumerate(drift_points):
+        fig.add_vline(
+            x=X[drift_point],
+            line_dash="dash",
+            line_color="red",
+            line_width=2,
+            annotation_text=f"Drift {i+1}",
+            annotation_position="top"
+        )
 
     # 레이아웃 설정
     title_map = {
@@ -77,11 +48,11 @@ def create_drift_visualization(X: np.ndarray, y: np.ndarray, drift_points: np.nd
             font=dict(size=20)
         ),
         xaxis_title="Time",
-        yaxis_title="Data distribution",
+        yaxis_title="Value",
         hovermode='closest',
         template='plotly_white',
-        height=400,
-        showlegend=(drift_type != "incremental"),
+        height=500,
+        showlegend=True,
         legend=dict(
             yanchor="top",
             y=0.99,
@@ -89,13 +60,14 @@ def create_drift_visualization(X: np.ndarray, y: np.ndarray, drift_points: np.nd
             x=0.99
         ),
         xaxis=dict(
-            showgrid=False,
-            showticklabels=True
+            showgrid=True,
+            gridwidth=1,
+            gridcolor='LightGray'
         ),
         yaxis=dict(
-            showgrid=False,
-            showticklabels=False,
-            range=[0.5, 1.5]
+            showgrid=True,
+            gridwidth=1,
+            gridcolor='LightGray'
         ),
         plot_bgcolor='white'
     )
@@ -121,57 +93,31 @@ def create_comparison_visualization(drift_data_dict: dict) -> go.Figure:
         if drift_type in drift_data_dict:
             X, y, drift_points = drift_data_dict[drift_type]
 
-            if drift_type == "incremental":
-                # Incremental drift: 연속 색상 변화
-                colors = []
-                for val in y:
-                    r = int(50 + 50 * val)
-                    g = int(100 + 79 * val)
-                    b = int(200 - 87 * val)
-                    colors.append(f'rgb({r}, {g}, {b})')
-
-                fig.add_trace(
-                    go.Scatter(
-                        x=X,
-                        y=np.ones(len(X)),
-                        mode='markers',
-                        marker=dict(color=colors, size=5, symbol='square', line=dict(width=0)),
-                        showlegend=False
-                    ),
+            # 라인 그래프 추가
+            fig.add_trace(
+                go.Scatter(
+                    x=X,
+                    y=y,
+                    mode='lines',
+                    line=dict(color='rgb(100, 140, 200)', width=1),
+                    showlegend=False
+                ),
+                row=row, col=col
+            )
+
+            # 드리프트 지점 표시
+            for drift_point in drift_points:
+                fig.add_vline(
+                    x=X[drift_point],
+                    line_dash="dash",
+                    line_color="red",
+                    line_width=1,
                     row=row, col=col
                 )
-            else:
-                # 이진 분류
-                class_0_mask = y == 0
-                class_1_mask = y == 1
-
-                if np.any(class_0_mask):
-                    fig.add_trace(
-                        go.Scatter(
-                            x=X[class_0_mask],
-                            y=np.ones(np.sum(class_0_mask)),
-                            mode='markers',
-                            marker=dict(color='rgb(65, 105, 225)', size=5, symbol='square', line=dict(width=0)),
-                            showlegend=False
-                        ),
-                        row=row, col=col
-                    )
-
-                if np.any(class_1_mask):
-                    fig.add_trace(
-                        go.Scatter(
-                            x=X[class_1_mask],
-                            y=np.ones(np.sum(class_1_mask)),
-                            mode='markers',
-                            marker=dict(color='rgb(50, 205, 50)', size=5, symbol='square', line=dict(width=0)),
-                            showlegend=False
-                        ),
-                        row=row, col=col
-                    )
 
     # 레이아웃 설정
-    fig.update_xaxes(title_text="Time", showgrid=False, showticklabels=False)
-    fig.update_yaxes(title_text="Data distribution", showgrid=False, showticklabels=False, range=[0.5, 1.5])
+    fig.update_xaxes(title_text="Time", showgrid=True, gridcolor='LightGray')
+    fig.update_yaxes(title_text="Value", showgrid=True, gridcolor='LightGray')
     fig.update_layout(
         height=800,
         title_text="Concept Drift Types Comparison",