Initial commit: Django quantitative strategy executor

- Django 5.2.7 project with Python 3.13+ - Quant strategy management system with version control - Strategy implementations using registry pattern - API endpoints for strategy listing and execution - Sample strategy implementations (MovingAverage, RSI, BollingerBand) - Async strategy execution with status tracking 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
2025-10-04 13:50:46 +09:00
commit 01403c7df4
22 changed files with 1031 additions and 0 deletions
--- a/strategies/init.py
+++ b/strategies/init.py
--- a/strategies/admin.py
+++ b/strategies/admin.py
@@ -0,0 +1,3 @@
+from django.contrib import admin
+
+# Register your models here.
--- a/strategies/apps.py
+++ b/strategies/apps.py
@@ -0,0 +1,6 @@
+from django.apps import AppConfig
+
+
+class StrategiesConfig(AppConfig):
+    default_auto_field = 'django.db.models.BigAutoField'
+    name = 'strategies'
--- a/strategies/base.py
+++ b/strategies/base.py
@@ -0,0 +1,110 @@
+from abc import ABC, abstractmethod
+from typing import Dict, Any
+import json
+
+
+class BaseQuantStrategy(ABC):
+    """
+    퀀트 전략의 기본 클래스
+    모든 전략 구현체는 이 클래스를 상속받아야 합니다.
+    """
+
+    @property
+    @abstractmethod
+    def name(self) -> str:
+        """전략 이름"""
+        pass
+
+    @property
+    @abstractmethod
+    def description(self) -> str:
+        """전략 설명"""
+        pass
+
+    @property
+    @abstractmethod
+    def version(self) -> str:
+        """전략 버전"""
+        pass
+
+    @property
+    def default_parameters(self) -> Dict[str, Any]:
+        """기본 파라미터"""
+        return {}
+
+    @abstractmethod
+    def execute(self, parameters: Dict[str, Any] = None) -> Dict[str, Any]:
+        """
+        전략을 실행합니다.
+
+        Args:
+            parameters: 실행 파라미터
+
+        Returns:
+            실행 결과
+        """
+        pass
+
+    def validate_parameters(self, parameters: Dict[str, Any]) -> bool:
+        """
+        파라미터 유효성 검사
+
+        Args:
+            parameters: 검사할 파라미터
+
+        Returns:
+            유효성 검사 결과
+        """
+        return True
+
+
+class StrategyRegistry:
+    """전략 구현체 레지스트리"""
+
+    _strategies = {}
+
+    @classmethod
+    def register(cls, strategy_class: BaseQuantStrategy):
+        """전략 구현체를 레지스트리에 등록"""
+        strategy_instance = strategy_class()
+        key = f"{strategy_instance.name}:{strategy_instance.version}"
+        cls._strategies[key] = strategy_class
+        return strategy_class
+
+    @classmethod
+    def get_strategy(cls, name: str, version: str) -> BaseQuantStrategy:
+        """전략 구현체를 가져옵니다"""
+        key = f"{name}:{version}"
+        strategy_class = cls._strategies.get(key)
+        if strategy_class:
+            return strategy_class()
+        raise ValueError(f"Strategy {key} not found in registry")
+
+    @classmethod
+    def list_strategies(cls) -> Dict[str, Dict[str, Any]]:
+        """등록된 모든 전략 목록을 반환합니다"""
+        strategies = {}
+        for key, strategy_class in cls._strategies.items():
+            strategy_instance = strategy_class()
+            name = strategy_instance.name
+            if name not in strategies:
+                strategies[name] = {
+                    'name': name,
+                    'description': strategy_instance.description,
+                    'versions': []
+                }
+            strategies[name]['versions'].append({
+                'version': strategy_instance.version,
+                'default_parameters': strategy_instance.default_parameters
+            })
+        return strategies
+
+    @classmethod
+    def get_strategy_key(cls, name: str, version: str) -> str:
+        """전략 키를 생성합니다"""
+        return f"{name}:{version}"
+
+
+def strategy(cls):
+    """전략 구현체를 자동으로 레지스트리에 등록하는 데코레이터"""
+    return StrategyRegistry.register(cls)
--- a/strategies/implementations.py
+++ b/strategies/implementations.py
@@ -0,0 +1,202 @@
+from typing import Dict, Any
+import time
+import random
+import math
+from .base import BaseQuantStrategy, strategy
+
+
+@strategy
+class MovingAverageCrossover(BaseQuantStrategy):
+    """이동평균선 교차 전략"""
+
+    @property
+    def name(self) -> str:
+        return "MovingAverageCrossover"
+
+    @property
+    def description(self) -> str:
+        return "단기 이동평균선이 장기 이동평균선을 상향 돌파할 때 매수, 하향 돌파할 때 매도하는 전략"
+
+    @property
+    def version(self) -> str:
+        return "1.0.0"
+
+    @property
+    def default_parameters(self) -> Dict[str, Any]:
+        return {
+            "short_window": 20,
+            "long_window": 50,
+            "initial_capital": 100000,
+            "position_size": 0.1
+        }
+
+    def validate_parameters(self, parameters: Dict[str, Any]) -> bool:
+        required_params = ["short_window", "long_window", "initial_capital"]
+        for param in required_params:
+            if param not in parameters:
+                return False
+
+        if parameters["short_window"] >= parameters["long_window"]:
+            return False
+
+        return True
+
+    def execute(self, parameters: Dict[str, Any] = None) -> Dict[str, Any]:
+        if parameters is None:
+            parameters = self.default_parameters
+
+        if not self.validate_parameters(parameters):
+            raise ValueError("Invalid parameters")
+
+        # 시뮬레이션 실행
+        time.sleep(1)  # 실행 시간 시뮬레이션
+
+        # 모의 결과 생성
+        profit_rate = random.uniform(-0.15, 0.25)
+        trades_count = random.randint(15, 60)
+        win_rate = random.uniform(0.45, 0.75)
+
+        return {
+            "strategy": self.name,
+            "version": self.version,
+            "profit_loss": round(parameters["initial_capital"] * profit_rate, 2),
+            "profit_rate": round(profit_rate * 100, 2),
+            "trades_executed": trades_count,
+            "win_rate": round(win_rate, 3),
+            "execution_time": "1.2s",
+            "parameters_used": parameters,
+            "final_capital": round(parameters["initial_capital"] * (1 + profit_rate), 2)
+        }
+
+
+@strategy
+class RSIMeanReversion(BaseQuantStrategy):
+    """RSI 평균회귀 전략"""
+
+    @property
+    def name(self) -> str:
+        return "RSIMeanReversion"
+
+    @property
+    def description(self) -> str:
+        return "RSI 지표를 이용한 평균회귀 전략. RSI가 과매수/과매도 구간에서 반대 방향으로 거래"
+
+    @property
+    def version(self) -> str:
+        return "1.0.0"
+
+    @property
+    def default_parameters(self) -> Dict[str, Any]:
+        return {
+            "rsi_period": 14,
+            "oversold_threshold": 30,
+            "overbought_threshold": 70,
+            "initial_capital": 100000,
+            "position_size": 0.05
+        }
+
+    def validate_parameters(self, parameters: Dict[str, Any]) -> bool:
+        required_params = ["rsi_period", "oversold_threshold", "overbought_threshold", "initial_capital"]
+        for param in required_params:
+            if param not in parameters:
+                return False
+
+        if not (0 < parameters["oversold_threshold"] < parameters["overbought_threshold"] < 100):
+            return False
+
+        return True
+
+    def execute(self, parameters: Dict[str, Any] = None) -> Dict[str, Any]:
+        if parameters is None:
+            parameters = self.default_parameters
+
+        if not self.validate_parameters(parameters):
+            raise ValueError("Invalid parameters")
+
+        # 시뮬레이션 실행
+        time.sleep(1.5)  # 실행 시간 시뮬레이션
+
+        # 모의 결과 생성
+        profit_rate = random.uniform(-0.10, 0.18)
+        trades_count = random.randint(25, 80)
+        win_rate = random.uniform(0.40, 0.65)
+
+        return {
+            "strategy": self.name,
+            "version": self.version,
+            "profit_loss": round(parameters["initial_capital"] * profit_rate, 2),
+            "profit_rate": round(profit_rate * 100, 2),
+            "trades_executed": trades_count,
+            "win_rate": round(win_rate, 3),
+            "execution_time": "1.5s",
+            "parameters_used": parameters,
+            "final_capital": round(parameters["initial_capital"] * (1 + profit_rate), 2),
+            "max_drawdown": round(random.uniform(0.05, 0.20), 3)
+        }
+
+
+@strategy
+class BollingerBandBreakout(BaseQuantStrategy):
+    """볼린저 밴드 돌파 전략"""
+
+    @property
+    def name(self) -> str:
+        return "BollingerBandBreakout"
+
+    @property
+    def description(self) -> str:
+        return "볼린저 밴드 상한선 돌파시 매수, 하한선 돌파시 매도하는 돌파 전략"
+
+    @property
+    def version(self) -> str:
+        return "2.0.0"
+
+    @property
+    def default_parameters(self) -> Dict[str, Any]:
+        return {
+            "period": 20,
+            "std_dev": 2.0,
+            "initial_capital": 100000,
+            "position_size": 0.08,
+            "stop_loss": 0.05
+        }
+
+    def validate_parameters(self, parameters: Dict[str, Any]) -> bool:
+        required_params = ["period", "std_dev", "initial_capital"]
+        for param in required_params:
+            if param not in parameters:
+                return False
+
+        if parameters["std_dev"] <= 0 or parameters["period"] <= 0:
+            return False
+
+        return True
+
+    def execute(self, parameters: Dict[str, Any] = None) -> Dict[str, Any]:
+        if parameters is None:
+            parameters = self.default_parameters
+
+        if not self.validate_parameters(parameters):
+            raise ValueError("Invalid parameters")
+
+        # 시뮬레이션 실행
+        time.sleep(2)  # 실행 시간 시뮬레이션
+
+        # 모의 결과 생성
+        profit_rate = random.uniform(-0.20, 0.30)
+        trades_count = random.randint(10, 40)
+        win_rate = random.uniform(0.35, 0.70)
+
+        return {
+            "strategy": self.name,
+            "version": self.version,
+            "profit_loss": round(parameters["initial_capital"] * profit_rate, 2),
+            "profit_rate": round(profit_rate * 100, 2),
+            "trades_executed": trades_count,
+            "win_rate": round(win_rate, 3),
+            "execution_time": "2.0s",
+            "parameters_used": parameters,
+            "final_capital": round(parameters["initial_capital"] * (1 + profit_rate), 2),
+            "sharpe_ratio": round(random.uniform(0.5, 2.5), 2),
+            "volatility": round(random.uniform(0.15, 0.35), 3)
+        }
--- a/strategies/migrations/0001_initial.py
+++ b/strategies/migrations/0001_initial.py
@@ -0,0 +1,61 @@
+# Generated by Django 5.2.7 on 2025-10-04 04:39
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='QuantStrategy',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('name', models.CharField(max_length=100, unique=True)),
+                ('description', models.TextField(blank=True)),
+                ('created_at', models.DateTimeField(auto_now_add=True)),
+                ('updated_at', models.DateTimeField(auto_now=True)),
+                ('is_active', models.BooleanField(default=True)),
+            ],
+            options={
+                'verbose_name_plural': 'Quant Strategies',
+            },
+        ),
+        migrations.CreateModel(
+            name='StrategyVersion',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('version', models.CharField(max_length=20)),
+                ('code', models.TextField()),
+                ('parameters', models.JSONField(blank=True, default=dict)),
+                ('created_at', models.DateTimeField(auto_now_add=True)),
+                ('is_current', models.BooleanField(default=False)),
+                ('strategy', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='versions', to='strategies.quantstrategy')),
+            ],
+            options={
+                'ordering': ['-created_at'],
+                'unique_together': {('strategy', 'version')},
+            },
+        ),
+        migrations.CreateModel(
+            name='StrategyExecution',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('execution_parameters', models.JSONField(blank=True, default=dict)),
+                ('status', models.CharField(choices=[('pending', 'Pending'), ('running', 'Running'), ('completed', 'Completed'), ('failed', 'Failed')], default='pending', max_length=20)),
+                ('started_at', models.DateTimeField(auto_now_add=True)),
+                ('completed_at', models.DateTimeField(blank=True, null=True)),
+                ('result', models.JSONField(blank=True, null=True)),
+                ('error_message', models.TextField(blank=True)),
+                ('strategy_version', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='strategies.strategyversion')),
+            ],
+            options={
+                'ordering': ['-started_at'],
+            },
+        ),
+    ]
--- a/strategies/migrations/0002_remove_strategyversion_code_and_more.py
+++ b/strategies/migrations/0002_remove_strategyversion_code_and_more.py
@@ -0,0 +1,22 @@
+# Generated by Django 5.2.7 on 2025-10-04 04:44
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('strategies', '0001_initial'),
+    ]
+
+    operations = [
+        migrations.RemoveField(
+            model_name='strategyversion',
+            name='code',
+        ),
+        migrations.AddField(
+            model_name='strategyversion',
+            name='implementation_key',
+            field=models.CharField(default='', help_text='전략 구현체의 레지스트리 키', max_length=200),
+        ),
+    ]
--- a/strategies/migrations/init.py
+++ b/strategies/migrations/init.py
--- a/strategies/models.py
+++ b/strategies/models.py
@@ -0,0 +1,81 @@
+from django.db import models
+from django.core.exceptions import ValidationError
+import json
+from .base import StrategyRegistry
+
+
+class QuantStrategy(models.Model):
+    name = models.CharField(max_length=100, unique=True)
+    description = models.TextField(blank=True)
+    created_at = models.DateTimeField(auto_now_add=True)
+    updated_at = models.DateTimeField(auto_now=True)
+    is_active = models.BooleanField(default=True)
+
+    def __str__(self):
+        return self.name
+
+    class Meta:
+        verbose_name_plural = "Quant Strategies"
+
+
+class StrategyVersion(models.Model):
+    strategy = models.ForeignKey(QuantStrategy, on_delete=models.CASCADE, related_name='versions')
+    version = models.CharField(max_length=20)
+    implementation_key = models.CharField(max_length=200, default="", help_text="전략 구현체의 레지스트리 키")
+    parameters = models.JSONField(default=dict, blank=True)
+    created_at = models.DateTimeField(auto_now_add=True)
+    is_current = models.BooleanField(default=False)
+
+    def clean(self):
+        if self.is_current:
+            StrategyVersion.objects.filter(
+                strategy=self.strategy,
+                is_current=True
+            ).exclude(pk=self.pk).update(is_current=False)
+
+        if not self.implementation_key:
+            self.implementation_key = StrategyRegistry.get_strategy_key(
+                self.strategy.name, self.version
+            )
+
+    def save(self, *args, **kwargs):
+        self.clean()
+        super().save(*args, **kwargs)
+
+    def get_strategy_implementation(self):
+        """전략 구현체 인스턴스를 반환합니다"""
+        try:
+            name, version = self.implementation_key.split(':')
+            return StrategyRegistry.get_strategy(name, version)
+        except ValueError:
+            raise ValidationError(f"Invalid implementation key: {self.implementation_key}")
+
+    def __str__(self):
+        return f"{self.strategy.name} v{self.version}"
+
+    class Meta:
+        unique_together = ['strategy', 'version']
+        ordering = ['-created_at']
+
+
+class StrategyExecution(models.Model):
+    EXECUTION_STATUS = [
+        ('pending', 'Pending'),
+        ('running', 'Running'),
+        ('completed', 'Completed'),
+        ('failed', 'Failed'),
+    ]
+
+    strategy_version = models.ForeignKey(StrategyVersion, on_delete=models.CASCADE)
+    execution_parameters = models.JSONField(default=dict, blank=True)
+    status = models.CharField(max_length=20, choices=EXECUTION_STATUS, default='pending')
+    started_at = models.DateTimeField(auto_now_add=True)
+    completed_at = models.DateTimeField(null=True, blank=True)
+    result = models.JSONField(null=True, blank=True)
+    error_message = models.TextField(blank=True)
+
+    def __str__(self):
+        return f"Execution of {self.strategy_version} - {self.status}"
+
+    class Meta:
+        ordering = ['-started_at']
--- a/strategies/tests.py
+++ b/strategies/tests.py
@@ -0,0 +1,3 @@
+from django.test import TestCase
+
+# Create your tests here.
--- a/strategies/urls.py
+++ b/strategies/urls.py
@@ -0,0 +1,9 @@
+from django.urls import path
+from . import views
+
+urlpatterns = [
+    path('strategies/', views.list_strategies, name='list_strategies'),
+    path('strategies/execute/', views.execute_strategy, name='execute_strategy'),
+    path('strategies/implementations/', views.list_available_implementations, name='list_available_implementations'),
+    path('executions/<int:execution_id>/', views.execution_status, name='execution_status'),
+]
--- a/strategies/views.py
+++ b/strategies/views.py
@@ -0,0 +1,161 @@
+from django.shortcuts import render, get_object_or_404
+from django.http import JsonResponse
+from django.views.decorators.csrf import csrf_exempt
+from django.views.decorators.http import require_http_methods
+from django.utils import timezone
+import json
+import threading
+import time
+
+from .models import QuantStrategy, StrategyVersion, StrategyExecution
+from .base import StrategyRegistry
+from . import implementations  # 구현체들을 로드하여 레지스트리에 등록
+
+
+@require_http_methods(["GET"])
+def list_strategies(request):
+    # 레지스트리에서 사용 가능한 전략들을 가져옴
+    available_strategies = StrategyRegistry.list_strategies()
+
+    # DB에서 관리되는 전략들을 가져옴
+    strategies = QuantStrategy.objects.filter(is_active=True).prefetch_related('versions')
+
+    strategy_list = []
+    for strategy in strategies:
+        current_version = strategy.versions.filter(is_current=True).first()
+        strategy_data = {
+            'id': strategy.id,
+            'name': strategy.name,
+            'description': strategy.description,
+            'created_at': strategy.created_at.isoformat(),
+            'current_version': {
+                'id': current_version.id,
+                'version': current_version.version,
+                'implementation_key': current_version.implementation_key,
+                'parameters': current_version.parameters,
+                'created_at': current_version.created_at.isoformat()
+            } if current_version else None,
+            'total_versions': strategy.versions.count(),
+            'available_implementations': available_strategies.get(strategy.name, {}).get('versions', [])
+        }
+        strategy_list.append(strategy_data)
+
+    return JsonResponse({
+        'strategies': strategy_list,
+        'total_count': len(strategy_list),
+        'available_implementations': available_strategies
+    })
+
+
+@csrf_exempt
+@require_http_methods(["POST"])
+def execute_strategy(request):
+    try:
+        data = json.loads(request.body)
+        strategy_name = data.get('strategy_name')
+        version = data.get('version')
+        execution_parameters = data.get('parameters', {})
+
+        if not strategy_name:
+            return JsonResponse({
+                'error': 'strategy_name is required'
+            }, status=400)
+
+        strategy = get_object_or_404(QuantStrategy, name=strategy_name, is_active=True)
+
+        if version:
+            strategy_version = get_object_or_404(
+                StrategyVersion,
+                strategy=strategy,
+                version=version
+            )
+        else:
+            strategy_version = strategy.versions.filter(is_current=True).first()
+            if not strategy_version:
+                return JsonResponse({
+                    'error': 'No current version found for this strategy'
+                }, status=404)
+
+        execution = StrategyExecution.objects.create(
+            strategy_version=strategy_version,
+            execution_parameters=execution_parameters,
+            status='pending'
+        )
+
+        def run_strategy():
+            try:
+                execution.status = 'running'
+                execution.save()
+
+                # 전략 구현체 인스턴스를 가져와서 실행
+                strategy_impl = strategy_version.get_strategy_implementation()
+
+                # 기본 파라미터와 실행 파라미터를 병합
+                merged_parameters = strategy_impl.default_parameters.copy()
+                merged_parameters.update(execution_parameters)
+
+                # 전략 실행
+                result = strategy_impl.execute(merged_parameters)
+
+                execution.status = 'completed'
+                execution.result = result
+                execution.completed_at = timezone.now()
+                execution.save()
+
+            except Exception as e:
+                execution.status = 'failed'
+                execution.error_message = str(e)
+                execution.completed_at = timezone.now()
+                execution.save()
+
+        thread = threading.Thread(target=run_strategy)
+        thread.start()
+
+        return JsonResponse({
+            'execution_id': execution.id,
+            'status': 'pending',
+            'message': 'Strategy execution started'
+        })
+
+    except json.JSONDecodeError:
+        return JsonResponse({
+            'error': 'Invalid JSON in request body'
+        }, status=400)
+    except Exception as e:
+        return JsonResponse({
+            'error': str(e)
+        }, status=500)
+
+
+@require_http_methods(["GET"])
+def execution_status(request, execution_id):
+    execution = get_object_or_404(StrategyExecution, id=execution_id)
+
+    response_data = {
+        'execution_id': execution.id,
+        'strategy': execution.strategy_version.strategy.name,
+        'version': execution.strategy_version.version,
+        'status': execution.status,
+        'started_at': execution.started_at.isoformat(),
+        'execution_parameters': execution.execution_parameters
+    }
+
+    if execution.completed_at:
+        response_data['completed_at'] = execution.completed_at.isoformat()
+
+    if execution.status == 'completed' and execution.result:
+        response_data['result'] = execution.result
+
+    if execution.status == 'failed' and execution.error_message:
+        response_data['error_message'] = execution.error_message
+
+    return JsonResponse(response_data)
+
+
+@require_http_methods(["GET"])
+def list_available_implementations(request):
+    """레지스트리에 등록된 모든 전략 구현체 목록을 반환"""
+    available_strategies = StrategyRegistry.list_strategies()
+    return JsonResponse({
+        'available_implementations': available_strategies
+    })