INTRODUCTION
The intersection of Large Language Models (LLMs) and Domain Specific Languages (DSLs) represents a transformative approach to software development. This convergence enables the automatic generation of specialized programming languages tailored to specific business domains, either from natural language descriptions or formal Domain Driven Design (DDD) specifications. Such systems democratize the creation of domain-specific tools while maintaining the precision and expressiveness that DSLs provide.
A DSL generation system powered by LLMs serves as an intelligent intermediary that understands domain concepts expressed in human language or structured DDD models and translates them into executable domain-specific syntax. This capability addresses the traditional challenge of DSL development, which typically requires deep expertise in both the target domain and language design principles.
The fundamental premise underlying these systems is that LLMs, trained on vast corpora of code and documentation, can recognize patterns between domain descriptions and their corresponding linguistic representations. When combined with proper architectural patterns and validation mechanisms, this recognition capability can be harnessed to produce reliable and maintainable DSL implementations.
ARCHITECTURAL FOUNDATIONS
The architecture of an LLM-based DSL generation system comprises several interconnected components that work together to transform high-level specifications into executable domain languages. The core architecture follows a pipeline pattern where each stage refines and validates the transformation process.
The Input Processing Layer serves as the entry point for both natural language prompts and DDD specifications. This layer normalizes different input formats and extracts semantic information that will guide the generation process. For natural language inputs, this involves parsing intent, identifying domain entities, and extracting relationships. For DDD specifications, it involves parsing bounded contexts, aggregates, entities, and value objects.
class InputProcessor:
"""
Processes and normalizes different types of input specifications
for DSL generation.
"""
def __init__(self, nlp_pipeline, ddd_parser):
self.nlp_pipeline = nlp_pipeline
self.ddd_parser = ddd_parser
def process_natural_language(self, prompt):
"""
Extract domain concepts from natural language description.
"""
entities = self.nlp_pipeline.extract_entities(prompt)
relationships = self.nlp_pipeline.extract_relationships(prompt)
constraints = self.nlp_pipeline.extract_constraints(prompt)
return DomainModel(
entities=entities,
relationships=relationships,
constraints=constraints,
source_type="natural_language"
)
def process_ddd_specification(self, ddd_spec):
"""
Parse structured DDD specification into domain model.
"""
bounded_contexts = self.ddd_parser.parse_contexts(ddd_spec)
aggregates = self.ddd_parser.parse_aggregates(ddd_spec)
domain_events = self.ddd_parser.parse_events(ddd_spec)
return DomainModel(
bounded_contexts=bounded_contexts,
aggregates=aggregates,
domain_events=domain_events,
source_type="ddd_specification"
)
The Domain Model Abstraction Layer creates a unified representation of domain concepts regardless of input source. This abstraction enables the system to apply consistent generation logic while accommodating different specification formats. The domain model captures entities, their attributes, relationships, constraints, and behavioral patterns in a format optimized for LLM consumption.
The LLM Integration Layer manages communication with language models, whether local or remote. This layer implements prompt engineering strategies, manages context windows, handles token limitations, and provides fallback mechanisms. The integration supports multiple LLM providers and model types, allowing for flexible deployment scenarios.
class LLMIntegrator:
"""
Manages interaction with various LLM providers for DSL generation.
"""
def __init__(self, primary_llm, fallback_llm=None):
self.primary_llm = primary_llm
self.fallback_llm = fallback_llm
self.prompt_templates = self._load_prompt_templates()
def generate_dsl_syntax(self, domain_model, generation_context):
"""
Generate DSL syntax using the configured LLM.
"""
prompt = self._construct_generation_prompt(domain_model, generation_context)
try:
response = self.primary_llm.generate(
prompt=prompt,
max_tokens=2048,
temperature=0.2,
stop_sequences=["END_DSL"]
)
return self._parse_dsl_response(response)
except Exception as e:
if self.fallback_llm:
return self._generate_with_fallback(prompt, e)
raise DSLGenerationError(f"Failed to generate DSL: {e}")
def _construct_generation_prompt(self, domain_model, context):
"""
Build a comprehensive prompt for DSL generation.
"""
template = self.prompt_templates["dsl_generation"]
return template.format(
domain_entities=domain_model.entities,
relationships=domain_model.relationships,
constraints=domain_model.constraints,
target_paradigm=context.paradigm,
syntax_preferences=context.syntax_preferences
)
The DSL Synthesis Engine coordinates the generation process by orchestrating interactions between the domain model, LLM integration layer, and validation components. This engine implements sophisticated prompt engineering techniques, manages generation iterations, and ensures consistency across generated language constructs.
The Validation and Refinement Layer ensures that generated DSLs meet quality standards and domain requirements. This layer performs syntactic validation, semantic consistency checks, and domain-specific constraint verification. When validation fails, the system can trigger refinement cycles that improve the generated output through iterative feedback.
NATURAL LANGUAGE TO DSL CONVERSION
Converting natural language descriptions into DSLs requires sophisticated understanding of both linguistic patterns and domain semantics. The process begins with intent recognition, where the system identifies the primary purpose and scope of the desired DSL. This involves analyzing the natural language input to extract key domain concepts, operational patterns, and structural requirements.
The semantic extraction phase employs named entity recognition and relationship extraction to identify domain-specific terminology and concepts. The system must distinguish between different types of entities such as business objects, processes, rules, and constraints. This extraction process is crucial because it forms the foundation for the subsequent DSL structure.
class NaturalLanguageDSLGenerator:
"""
Generates DSL from natural language descriptions using LLM capabilities.
"""
def __init__(self, llm_integrator, domain_analyzer):
self.llm_integrator = llm_integrator
self.domain_analyzer = domain_analyzer
def generate_from_description(self, description, target_domain):
"""
Convert natural language description to DSL specification.
"""
# Extract domain concepts from natural language
domain_analysis = self.domain_analyzer.analyze_description(description)
# Identify DSL patterns and structures
dsl_patterns = self._identify_dsl_patterns(domain_analysis)
# Generate syntax rules and grammar
syntax_specification = self._generate_syntax_specification(
domain_analysis, dsl_patterns, target_domain
)
# Create executable DSL implementation
dsl_implementation = self._synthesize_dsl_implementation(
syntax_specification, domain_analysis
)
return DSLArtifact(
specification=syntax_specification,
implementation=dsl_implementation,
metadata=self._generate_metadata(domain_analysis)
)
def _identify_dsl_patterns(self, domain_analysis):
"""
Identify appropriate DSL patterns based on domain characteristics.
"""
patterns = []
if domain_analysis.has_sequential_processes():
patterns.append("workflow_dsl")
if domain_analysis.has_rule_based_logic():
patterns.append("rule_engine_dsl")
if domain_analysis.has_data_transformations():
patterns.append("transformation_dsl")
return patterns
The context understanding component analyzes the broader context in which the DSL will operate. This includes identifying the target users, typical use cases, performance requirements, and integration constraints. Understanding context is essential for making appropriate design decisions about syntax complexity, abstraction levels, and feature priorities.
The iterative refinement process allows the system to improve DSL quality through multiple generation cycles. Initial generations often require refinement to address ambiguities, resolve conflicts, or incorporate additional requirements that emerge during the analysis phase. The system maintains conversation history to enable coherent refinement across multiple iterations.
DOMAIN DRIVEN DESIGN TO DSL CONVERSION
Converting DDD specifications to DSLs leverages the structured nature of DDD artifacts to create more precise and comprehensive domain languages. DDD provides a rich vocabulary of concepts including bounded contexts, aggregates, entities, value objects, domain services, and domain events that can be directly mapped to DSL constructs.
The bounded context analysis phase examines the DDD specification to identify distinct areas of the domain that require different linguistic representations. Each bounded context may necessitate its own DSL dialect or specialized constructs within a unified language. This analysis ensures that the generated DSL respects domain boundaries and maintains conceptual integrity.
class DDDToDSLConverter:
"""
Converts Domain Driven Design specifications into executable DSLs.
"""
def __init__(self, ddd_parser, llm_integrator, dsl_synthesizer):
self.ddd_parser = ddd_parser
self.llm_integrator = llm_integrator
self.dsl_synthesizer = dsl_synthesizer
def convert_specification(self, ddd_specification):
"""
Transform DDD specification into comprehensive DSL.
"""
# Parse DDD artifacts
parsed_ddd = self.ddd_parser.parse_complete_specification(ddd_specification)
# Map DDD concepts to DSL constructs
dsl_mapping = self._create_concept_mapping(parsed_ddd)
# Generate syntax for each domain concept
syntax_elements = self._generate_syntax_elements(dsl_mapping)
# Synthesize complete DSL grammar
complete_grammar = self.dsl_synthesizer.synthesize_grammar(syntax_elements)
# Generate implementation artifacts
implementation = self._generate_implementation_artifacts(
complete_grammar, parsed_ddd
)
return DSLArtifact(
grammar=complete_grammar,
implementation=implementation,
ddd_mapping=dsl_mapping,
validation_rules=self._generate_validation_rules(parsed_ddd)
)
def _create_concept_mapping(self, parsed_ddd):
"""
Create mapping between DDD concepts and DSL language constructs.
"""
mapping = ConceptMapping()
# Map aggregates to DSL entities
for aggregate in parsed_ddd.aggregates:
mapping.add_entity_mapping(
ddd_concept=aggregate,
dsl_construct=self._design_aggregate_syntax(aggregate)
)
# Map domain events to DSL event constructs
for event in parsed_ddd.domain_events:
mapping.add_event_mapping(
ddd_concept=event,
dsl_construct=self._design_event_syntax(event)
)
# Map domain services to DSL operations
for service in parsed_ddd.domain_services:
mapping.add_operation_mapping(
ddd_concept=service,
dsl_construct=self._design_service_syntax(service)
)
return mapping
The aggregate modeling component translates DDD aggregates into DSL entity definitions. Aggregates represent consistency boundaries and encapsulate business logic, making them natural candidates for DSL entity types. The conversion process preserves aggregate invariants and business rules while creating intuitive syntax for aggregate manipulation.
The domain event mapping transforms DDD domain events into DSL event constructs that support event-driven programming patterns. This mapping ensures that the generated DSL can express complex event flows and maintain the temporal semantics inherent in domain events.
The ubiquitous language preservation ensures that the generated DSL maintains the terminology and concepts established in the DDD process. This preservation is crucial for maintaining alignment between the technical implementation and business understanding of the domain.
LOCAL VERSUS REMOTE LLM CONSIDERATIONS
The choice between local and remote LLM deployment significantly impacts system architecture, performance characteristics, and operational requirements. Each approach presents distinct advantages and challenges that must be carefully evaluated based on specific use case requirements.
Local LLM deployment provides complete control over the inference environment and eliminates external dependencies. Local models can be fine-tuned specifically for DSL generation tasks, potentially improving accuracy and reducing hallucination rates. The deployment also ensures data privacy since no information leaves the local environment during processing.
class LocalLLMProvider:
"""
Manages local LLM deployment for DSL generation.
"""
def __init__(self, model_path, device_config):
self.model_path = model_path
self.device_config = device_config
self.model = None
self.tokenizer = None
def initialize_model(self):
"""
Load and initialize the local LLM for inference.
"""
try:
self.tokenizer = AutoTokenizer.from_pretrained(self.model_path)
self.model = AutoModelForCausalLM.from_pretrained(
self.model_path,
torch_dtype=torch.float16,
device_map=self.device_config.device_map,
trust_remote_code=True
)
# Optimize for inference
self.model.eval()
if self.device_config.use_compilation:
self.model = torch.compile(self.model)
except Exception as e:
raise LocalLLMInitializationError(f"Failed to initialize local LLM: {e}")
def generate_response(self, prompt, generation_config):
"""
Generate response using local LLM with specified configuration.
"""
if not self.model:
raise RuntimeError("Model not initialized")
inputs = self.tokenizer.encode(prompt, return_tensors="pt")
with torch.no_grad():
outputs = self.model.generate(
inputs,
max_length=generation_config.max_length,
temperature=generation_config.temperature,
do_sample=generation_config.do_sample,
pad_token_id=self.tokenizer.eos_token_id
)
response = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
return response[len(prompt):].strip()
However, local deployment requires significant computational resources and expertise in model management. The hardware requirements for running capable LLMs can be substantial, particularly for models that demonstrate strong performance on complex reasoning tasks. Additionally, local deployment necessitates ongoing maintenance, updates, and monitoring that may strain organizational resources.
Remote LLM services offer immediate access to state-of-the-art models without infrastructure investment. These services typically provide superior performance for complex tasks and benefit from continuous improvements and updates managed by specialized providers. The operational overhead is minimal, allowing development teams to focus on application logic rather than model management.
class RemoteLLMProvider:
"""
Manages remote LLM API integration for DSL generation.
"""
def __init__(self, api_config, rate_limiter):
self.api_config = api_config
self.rate_limiter = rate_limiter
self.client = self._initialize_client()
def _initialize_client(self):
"""
Initialize API client with proper authentication and configuration.
"""
return OpenAIClient(
api_key=self.api_config.api_key,
base_url=self.api_config.base_url,
timeout=self.api_config.timeout,
max_retries=self.api_config.max_retries
)
def generate_response(self, prompt, generation_config):
"""
Generate response using remote LLM API with rate limiting.
"""
# Apply rate limiting
self.rate_limiter.acquire()
try:
response = self.client.chat.completions.create(
model=generation_config.model_name,
messages=[{"role": "user", "content": prompt}],
max_tokens=generation_config.max_tokens,
temperature=generation_config.temperature,
top_p=generation_config.top_p,
frequency_penalty=generation_config.frequency_penalty
)
return response.choices[0].message.content
except APIError as e:
raise RemoteLLMError(f"API request failed: {e}")
except RateLimitError as e:
# Implement exponential backoff
self._handle_rate_limit(e)
return self.generate_response(prompt, generation_config)
finally:
self.rate_limiter.release()
The primary concerns with remote services include data privacy, network dependency, and ongoing costs. Sensitive domain information must traverse external networks, potentially creating compliance challenges. Network connectivity issues can disrupt the generation process, and usage-based pricing models may become expensive for high-volume applications.
The optimal approach often involves a hybrid strategy that leverages both local and remote capabilities. Critical or sensitive operations can utilize local models while complex generation tasks benefit from remote services. This hybrid approach provides flexibility and resilience while optimizing for both performance and cost considerations.
IMPLEMENTATION ARCHITECTURE DETAILS
The implementation architecture must address several critical concerns including prompt engineering, context management, error handling, and result validation. The prompt engineering component is particularly crucial as it directly influences the quality and consistency of generated DSLs.
Effective prompt engineering for DSL generation requires careful construction of prompts that provide sufficient context while maintaining clarity and focus. The prompts must include domain information, syntax preferences, target use cases, and quality constraints. Additionally, the prompts should incorporate examples of well-formed DSL constructs to guide the generation process.
class PromptEngineer:
"""
Manages sophisticated prompt construction for DSL generation tasks.
"""
def __init__(self, template_repository, example_database):
self.template_repository = template_repository
self.example_database = example_database
def construct_dsl_generation_prompt(self, domain_model, generation_context):
"""
Build comprehensive prompt for DSL generation with examples and constraints.
"""
base_template = self.template_repository.get_template("dsl_generation_base")
# Select relevant examples based on domain characteristics
relevant_examples = self.example_database.find_similar_examples(
domain_model.characteristics,
limit=3
)
# Construct constraint specifications
constraint_spec = self._build_constraint_specification(generation_context)
# Assemble complete prompt
complete_prompt = base_template.format(
domain_description=self._serialize_domain_model(domain_model),
syntax_examples=self._format_examples(relevant_examples),
generation_constraints=constraint_spec,
target_paradigm=generation_context.paradigm,
quality_requirements=generation_context.quality_requirements
)
return complete_prompt
def _build_constraint_specification(self, context):
"""
Create detailed constraint specification for generation guidance.
"""
constraints = []
if context.syntax_style:
constraints.append(f"Syntax style: {context.syntax_style}")
if context.complexity_limit:
constraints.append(f"Maximum complexity: {context.complexity_limit}")
if context.reserved_keywords:
constraints.append(f"Avoid keywords: {', '.join(context.reserved_keywords)}")
return "\n".join(constraints)
The context management system maintains conversation state and generation history to enable coherent multi-turn interactions. This system tracks previous generation attempts, user feedback, and refinement requests to ensure that subsequent generations build upon previous work rather than starting from scratch.
The error handling and recovery mechanisms address the inherent unpredictability of LLM outputs. These mechanisms include syntax validation, semantic consistency checking, and automatic retry logic with modified prompts when generation fails. The system must gracefully handle partial failures and provide meaningful feedback to users.
The result validation framework ensures that generated DSLs meet quality standards and functional requirements. This framework performs multiple levels of validation including syntactic correctness, semantic consistency, domain alignment, and usability assessment. Failed validations trigger refinement cycles that attempt to address identified issues.
class DSLValidator:
"""
Comprehensive validation framework for generated DSL artifacts.
"""
def __init__(self, syntax_validator, semantic_validator, domain_validator):
self.syntax_validator = syntax_validator
self.semantic_validator = semantic_validator
self.domain_validator = domain_validator
def validate_generated_dsl(self, dsl_artifact, domain_model):
"""
Perform comprehensive validation of generated DSL.
"""
validation_results = ValidationResults()
# Syntactic validation
syntax_result = self.syntax_validator.validate_syntax(
dsl_artifact.grammar,
dsl_artifact.implementation
)
validation_results.add_syntax_result(syntax_result)
# Semantic consistency validation
semantic_result = self.semantic_validator.validate_semantics(
dsl_artifact,
domain_model
)
validation_results.add_semantic_result(semantic_result)
# Domain alignment validation
domain_result = self.domain_validator.validate_domain_alignment(
dsl_artifact,
domain_model
)
validation_results.add_domain_result(domain_result)
# Generate validation report
validation_report = self._generate_validation_report(validation_results)
return ValidationOutcome(
is_valid=validation_results.is_completely_valid(),
results=validation_results,
report=validation_report,
suggested_improvements=self._suggest_improvements(validation_results)
)
def _suggest_improvements(self, validation_results):
"""
Generate specific improvement suggestions based on validation failures.
"""
suggestions = []
if validation_results.has_syntax_errors():
suggestions.extend(self._generate_syntax_suggestions(validation_results))
if validation_results.has_semantic_issues():
suggestions.extend(self._generate_semantic_suggestions(validation_results))
if validation_results.has_domain_misalignment():
suggestions.extend(self._generate_domain_suggestions(validation_results))
return suggestions
BENEFITS AND ADVANTAGES
LLM-based DSL generation systems provide numerous benefits that address traditional challenges in domain-specific language development. The primary advantage is the dramatic reduction in development time and expertise requirements for creating domain-specific languages. Traditional DSL development requires deep expertise in both the target domain and language design principles, creating a significant barrier to adoption.
The accessibility improvement enables domain experts without extensive programming backgrounds to participate directly in DSL creation. By expressing requirements in natural language or structured domain models, subject matter experts can contribute to language design without requiring translation through technical intermediaries. This direct participation improves the alignment between business needs and technical implementation.
The rapid prototyping capability allows organizations to experiment with different DSL approaches quickly and cost-effectively. Traditional DSL development involves significant upfront investment before the utility of the approach can be evaluated. LLM-based generation enables rapid creation of prototype DSLs that can be tested and refined based on actual usage experience.
The consistency and standardization benefits emerge from the systematic approach to DSL generation. LLMs can apply consistent design patterns and naming conventions across different domain areas, creating a more coherent ecosystem of domain-specific languages within an organization. This consistency reduces learning overhead and improves maintainability.
The evolutionary capability enables DSLs to adapt and improve over time based on usage patterns and changing requirements. The generation system can incorporate feedback and refinement requests to produce updated versions of DSLs that better serve evolving needs. This evolutionary approach contrasts with traditional DSL development where modifications require significant manual effort.
DISADVANTAGES AND LIMITATIONS
Despite significant benefits, LLM-based DSL generation systems face several important limitations that must be carefully considered. The quality consistency challenge represents a fundamental concern as LLM outputs can vary significantly between generation attempts. Unlike traditional deterministic development processes, LLM-based systems may produce different results for identical inputs, creating uncertainty about output quality.
The domain expertise requirement remains significant despite the accessibility improvements. While LLMs can assist with language design, they cannot replace deep domain understanding required for creating truly effective DSLs. The generated languages may lack subtle domain-specific optimizations or fail to capture important edge cases that domain experts would naturally consider.
The validation complexity increases substantially when using LLM-generated artifacts. Traditional software development relies on well-established testing and validation methodologies, but validating generated DSLs requires new approaches that can assess both syntactic correctness and semantic appropriateness. This validation challenge becomes particularly acute for mission-critical applications.
The maintenance and evolution challenges emerge as generated DSLs require ongoing support and enhancement. While the initial generation may be rapid, maintaining and evolving LLM-generated languages requires careful coordination between automated generation capabilities and manual refinement processes. Organizations must develop new workflows and expertise to manage this hybrid development approach.
The dependency and risk considerations include reliance on external LLM services, potential model obsolescence, and the need for specialized expertise in prompt engineering and LLM management. These dependencies create new categories of technical risk that organizations must assess and mitigate.
BEST PRACTICES AND RECOMMENDATIONS
Successful implementation of LLM-based DSL generation systems requires adherence to several critical best practices that address the unique challenges of this approach. The iterative development methodology proves essential for achieving high-quality results. Rather than attempting to generate complete DSLs in single iterations, successful implementations employ multiple refinement cycles that progressively improve quality and completeness.
The validation-driven approach ensures that quality standards are maintained throughout the generation process. This approach involves implementing comprehensive validation frameworks that assess multiple dimensions of DSL quality including syntactic correctness, semantic consistency, domain alignment, and usability characteristics. Validation should be automated wherever possible to enable rapid feedback cycles.
The hybrid expertise model combines automated generation capabilities with human domain expertise to achieve optimal results. This model recognizes that LLMs excel at pattern recognition and syntax generation while humans provide domain insight and quality assessment. Successful implementations establish clear roles and workflows that leverage the strengths of both automated and human capabilities.
The prompt engineering discipline requires systematic development and maintenance of prompt templates, examples, and constraints that guide the generation process. Organizations should invest in building comprehensive prompt libraries that capture domain-specific knowledge and generation preferences. These libraries should be version-controlled and continuously refined based on generation experience.
The documentation and traceability practices ensure that generated DSLs can be understood, maintained, and evolved over time. This includes maintaining clear documentation of generation parameters, domain models, validation results, and refinement history. Traceability enables teams to understand how specific DSL features relate to domain requirements and generation decisions.
OPTIMAL DEPLOYMENT STRATEGIES
The optimal deployment strategy for LLM-based DSL generation systems depends on organizational requirements, technical constraints, and risk tolerance. For organizations with strong privacy requirements and sufficient technical resources, local LLM deployment provides maximum control and security. This approach requires investment in specialized hardware and expertise but eliminates external dependencies and data privacy concerns.
Organizations prioritizing rapid deployment and access to cutting-edge capabilities should consider remote LLM services as the primary approach. This strategy minimizes infrastructure requirements and provides immediate access to state-of-the-art models. However, it requires careful attention to data privacy, cost management, and service reliability considerations.
The hybrid deployment approach often provides the optimal balance of capabilities and constraints. This approach uses local models for sensitive or routine generation tasks while leveraging remote services for complex or specialized requirements. The hybrid strategy provides flexibility and resilience while optimizing for both performance and cost considerations.
The progressive deployment methodology enables organizations to start with simple use cases and gradually expand to more complex scenarios. This approach allows teams to develop expertise and refine processes before tackling mission-critical applications. Progressive deployment also enables learning and adaptation that improves subsequent implementations.
RUNNING EXAMPLE OVERVIEW
Throughout this article, we have referenced a comprehensive running example that demonstrates the implementation of an LLM-based DSL generation system for financial trading rules. This example illustrates the conversion of both natural language descriptions and DDD specifications into executable trading DSLs that can express complex financial logic in domain-appropriate syntax.
The example system supports multiple input formats including natural language descriptions of trading strategies and formal DDD specifications of financial domain models. The generated DSLs enable traders and quantitative analysts to express complex trading logic using familiar financial terminology while maintaining the precision required for automated execution.
The implementation demonstrates key architectural patterns including modular component design, comprehensive validation frameworks, and hybrid LLM deployment strategies. The example includes complete error handling, logging, and monitoring capabilities that would be required for production deployment.
CONCLUSION
LLM-based DSL generation represents a significant advancement in making domain-specific languages more accessible and practical for real-world applications. While challenges remain in areas such as quality consistency and validation complexity, the benefits of reduced development time, improved accessibility, and enhanced domain alignment make this approach compelling for many use cases.
Success with these systems requires careful attention to architectural design, validation frameworks, and deployment strategies. Organizations must invest in developing appropriate expertise and processes while maintaining realistic expectations about current capabilities and limitations.
The future evolution of this field will likely address current limitations through improved model capabilities, better validation techniques, and more sophisticated integration patterns. Organizations that begin experimenting with these approaches now will be well-positioned to benefit from future advances while developing valuable expertise in this emerging area.
COMPLETE RUNNING EXAMPLE
#!/usr/bin/env python3
"""
Complete LLM-based DSL Generation System for Financial Trading Rules
This implementation demonstrates a comprehensive system that generates
domain-specific languages for financial trading from both natural language
descriptions and Domain Driven Design specifications.
Author: System Architecture Team
Version: 1.0.0
License: MIT
"""
import json
import logging
import re
import time
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import Enum
from typing import Dict, List, Optional, Union, Any
from datetime import datetime
import asyncio
import aiohttp
import torch
from transformers import AutoTokenizer, AutoModelForCausalLM
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class DSLGenerationError(Exception):
"""Base exception for DSL generation errors."""
pass
class ValidationError(DSLGenerationError):
"""Exception raised when DSL validation fails."""
pass
class LLMError(DSLGenerationError):
"""Exception raised when LLM operations fail."""
pass
@dataclass
class DomainEntity:
"""Represents a domain entity extracted from specifications."""
name: str
attributes: Dict[str, str]
constraints: List[str] = field(default_factory=list)
relationships: List[str] = field(default_factory=list)
@dataclass
class DomainModel:
"""Unified representation of domain concepts."""
entities: List[DomainEntity] = field(default_factory=list)
relationships: List[str] = field(default_factory=list)
constraints: List[str] = field(default_factory=list)
bounded_contexts: List[str] = field(default_factory=list)
domain_events: List[str] = field(default_factory=list)
source_type: str = "unknown"
def to_dict(self) -> Dict[str, Any]:
"""Convert domain model to dictionary representation."""
return {
'entities': [
{
'name': entity.name,
'attributes': entity.attributes,
'constraints': entity.constraints,
'relationships': entity.relationships
}
for entity in self.entities
],
'relationships': self.relationships,
'constraints': self.constraints,
'bounded_contexts': self.bounded_contexts,
'domain_events': self.domain_events,
'source_type': self.source_type
}
@dataclass
class DSLArtifact:
"""Complete DSL generation result."""
grammar: str
implementation: str
metadata: Dict[str, Any]
validation_results: Optional['ValidationResults'] = None
generation_timestamp: datetime = field(default_factory=datetime.now)
def to_dict(self) -> Dict[str, Any]:
"""Convert DSL artifact to dictionary representation."""
return {
'grammar': self.grammar,
'implementation': self.implementation,
'metadata': self.metadata,
'generation_timestamp': self.generation_timestamp.isoformat(),
'validation_results': self.validation_results.to_dict() if self.validation_results else None
}
@dataclass
class ValidationResults:
"""Comprehensive validation results for generated DSL."""
syntax_valid: bool = True
semantic_valid: bool = True
domain_aligned: bool = True
syntax_errors: List[str] = field(default_factory=list)
semantic_warnings: List[str] = field(default_factory=list)
domain_issues: List[str] = field(default_factory=list)
suggestions: List[str] = field(default_factory=list)
def is_completely_valid(self) -> bool:
"""Check if all validation aspects pass."""
return self.syntax_valid and self.semantic_valid and self.domain_aligned
def to_dict(self) -> Dict[str, Any]:
"""Convert validation results to dictionary."""
return {
'syntax_valid': self.syntax_valid,
'semantic_valid': self.semantic_valid,
'domain_aligned': self.domain_aligned,
'syntax_errors': self.syntax_errors,
'semantic_warnings': self.semantic_warnings,
'domain_issues': self.domain_issues,
'suggestions': self.suggestions
}
class LLMProvider(ABC):
"""Abstract base class for LLM providers."""
@abstractmethod
async def generate(self, prompt: str, **kwargs) -> str:
"""Generate response from LLM."""
pass
@abstractmethod
def is_available(self) -> bool:
"""Check if LLM provider is available."""
pass
class LocalLLMProvider(LLMProvider):
"""Local LLM provider using Hugging Face transformers."""
def __init__(self, model_name: str, device: str = "auto"):
self.model_name = model_name
self.device = device
self.model = None
self.tokenizer = None
self._initialize_model()
def _initialize_model(self):
"""Initialize the local model and tokenizer."""
try:
logger.info(f"Initializing local model: {self.model_name}")
self.tokenizer = AutoTokenizer.from_pretrained(self.model_name)
self.model = AutoModelForCausalLM.from_pretrained(
self.model_name,
torch_dtype=torch.float16,
device_map=self.device,
trust_remote_code=True
)
self.model.eval()
logger.info("Local model initialized successfully")
except Exception as e:
logger.error(f"Failed to initialize local model: {e}")
raise LLMError(f"Local model initialization failed: {e}")
async def generate(self, prompt: str, max_tokens: int = 1024,
temperature: float = 0.7, **kwargs) -> str:
"""Generate response using local model."""
if not self.model or not self.tokenizer:
raise LLMError("Model not properly initialized")
try:
inputs = self.tokenizer.encode(prompt, return_tensors="pt")
with torch.no_grad():
outputs = self.model.generate(
inputs,
max_length=len(inputs[0]) + max_tokens,
temperature=temperature,
do_sample=True,
pad_token_id=self.tokenizer.eos_token_id,
**kwargs
)
response = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
return response[len(prompt):].strip()
except Exception as e:
logger.error(f"Local generation failed: {e}")
raise LLMError(f"Local generation failed: {e}")
def is_available(self) -> bool:
"""Check if local model is available."""
return self.model is not None and self.tokenizer is not None
class RemoteLLMProvider(LLMProvider):
"""Remote LLM provider using API services."""
def __init__(self, api_url: str, api_key: str, model_name: str):
self.api_url = api_url
self.api_key = api_key
self.model_name = model_name
self.session = None
async def _ensure_session(self):
"""Ensure aiohttp session is available."""
if not self.session:
self.session = aiohttp.ClientSession(
headers={"Authorization": f"Bearer {self.api_key}"}
)
async def generate(self, prompt: str, max_tokens: int = 1024,
temperature: float = 0.7, **kwargs) -> str:
"""Generate response using remote API."""
await self._ensure_session()
payload = {
"model": self.model_name,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": max_tokens,
"temperature": temperature,
**kwargs
}
try:
async with self.session.post(
f"{self.api_url}/chat/completions",
json=payload
) as response:
if response.status != 200:
error_text = await response.text()
raise LLMError(f"API request failed: {response.status} - {error_text}")
result = await response.json()
return result["choices"][0]["message"]["content"]
except aiohttp.ClientError as e:
logger.error(f"Remote generation failed: {e}")
raise LLMError(f"Remote generation failed: {e}")
def is_available(self) -> bool:
"""Check if remote API is available."""
# Simple availability check - in production, implement proper health check
return bool(self.api_url and self.api_key)
async def close(self):
"""Close the aiohttp session."""
if self.session:
await self.session.close()
class InputProcessor:
"""Processes and normalizes different types of input specifications."""
def __init__(self):
self.entity_patterns = [
r'\b([A-Z][a-z]+(?:[A-Z][a-z]+)*)\b', # PascalCase entities
r'\b(order|trade|position|portfolio|strategy|rule)\b', # Financial entities
]
self.relationship_patterns = [
r'(\w+)\s+(?:has|contains|includes|owns)\s+(\w+)',
r'(\w+)\s+(?:is|belongs to|part of)\s+(\w+)',
]
def process_natural_language(self, description: str) -> DomainModel:
"""Extract domain concepts from natural language description."""
logger.info("Processing natural language input")
entities = self._extract_entities(description)
relationships = self._extract_relationships(description)
constraints = self._extract_constraints(description)
return DomainModel(
entities=entities,
relationships=relationships,
constraints=constraints,
source_type="natural_language"
)
def process_ddd_specification(self, ddd_spec: Dict[str, Any]) -> DomainModel:
"""Parse structured DDD specification into domain model."""
logger.info("Processing DDD specification")
entities = []
for entity_spec in ddd_spec.get('entities', []):
entity = DomainEntity(
name=entity_spec['name'],
attributes=entity_spec.get('attributes', {}),
constraints=entity_spec.get('constraints', []),
relationships=entity_spec.get('relationships', [])
)
entities.append(entity)
return DomainModel(
entities=entities,
relationships=ddd_spec.get('relationships', []),
constraints=ddd_spec.get('constraints', []),
bounded_contexts=ddd_spec.get('bounded_contexts', []),
domain_events=ddd_spec.get('domain_events', []),
source_type="ddd_specification"
)
def _extract_entities(self, text: str) -> List[DomainEntity]:
"""Extract domain entities from text."""
entities = []
found_names = set()
for pattern in self.entity_patterns:
matches = re.findall(pattern, text, re.IGNORECASE)
for match in matches:
name = match.lower()
if name not in found_names:
found_names.add(name)
entities.append(DomainEntity(
name=name.capitalize(),
attributes=self._infer_attributes(name, text)
))
return entities
def _extract_relationships(self, text: str) -> List[str]:
"""Extract relationships from text."""
relationships = []
for pattern in self.relationship_patterns:
matches = re.findall(pattern, text, re.IGNORECASE)
for match in matches:
relationship = f"{match[0]} -> {match[1]}"
relationships.append(relationship)
return relationships
def _extract_constraints(self, text: str) -> List[str]:
"""Extract constraints from text."""
constraint_indicators = [
'must', 'should', 'cannot', 'required', 'mandatory',
'optional', 'minimum', 'maximum', 'between', 'greater than',
'less than', 'equal to'
]
constraints = []
sentences = text.split('.')
for sentence in sentences:
for indicator in constraint_indicators:
if indicator in sentence.lower():
constraints.append(sentence.strip())
break
return constraints
def _infer_attributes(self, entity_name: str, context: str) -> Dict[str, str]:
"""Infer likely attributes for an entity based on context."""
# Financial domain-specific attribute inference
financial_attributes = {
'order': {'symbol': 'string', 'quantity': 'number', 'price': 'number', 'side': 'string'},
'trade': {'symbol': 'string', 'quantity': 'number', 'price': 'number', 'timestamp': 'datetime'},
'position': {'symbol': 'string', 'quantity': 'number', 'average_price': 'number'},
'portfolio': {'name': 'string', 'value': 'number', 'positions': 'list'},
'strategy': {'name': 'string', 'parameters': 'dict', 'active': 'boolean'},
'rule': {'name': 'string', 'condition': 'string', 'action': 'string'}
}
return financial_attributes.get(entity_name.lower(), {'id': 'string', 'name': 'string'})
class PromptTemplateManager:
"""Manages prompt templates for DSL generation."""
def __init__(self):
self.templates = {
'dsl_generation': """
You are an expert in creating Domain Specific Languages (DSLs) for financial trading systems.
Domain Model:
{domain_model}
Requirements:
- Create a clean, readable DSL syntax for financial trading rules
- Include support for conditions, actions, and data references
- Use financial domain terminology
- Ensure the syntax is both human-readable and machine-parseable
- Include proper error handling constructs
Generate a complete DSL specification including:
1. Grammar definition in EBNF format
2. Python implementation with parser and interpreter
3. Example usage demonstrating key features
DSL Specification:
""",
'refinement': """
The following DSL has validation issues:
Original DSL:
{original_dsl}
Validation Issues:
{validation_issues}
Please refine the DSL to address these issues while maintaining the core functionality:
""",
'natural_language_analysis': """
Analyze the following natural language description for DSL generation:
Description: {description}
Extract and identify:
1. Key domain entities and their relationships
2. Business rules and constraints
3. Required operations and actions
4. Data types and structures needed
Analysis:
"""
}
def get_template(self, template_name: str) -> str:
"""Get a prompt template by name."""
return self.templates.get(template_name, "")
def format_template(self, template_name: str, **kwargs) -> str:
"""Format a template with provided parameters."""
template = self.get_template(template_name)
return template.format(**kwargs)
class DSLValidator:
"""Comprehensive validation framework for generated DSL artifacts."""
def __init__(self):
self.syntax_patterns = {
'balanced_brackets': r'[\[\]{}()]',
'valid_identifiers': r'\b[a-zA-Z_][a-zA-Z0-9_]*\b',
'string_literals': r'"[^"]*"|\'[^\']*\'',
}
def validate_dsl(self, dsl_artifact: DSLArtifact, domain_model: DomainModel) -> ValidationResults:
"""Perform comprehensive validation of generated DSL."""
logger.info("Starting DSL validation")
results = ValidationResults()
# Syntax validation
self._validate_syntax(dsl_artifact, results)
# Semantic validation
self._validate_semantics(dsl_artifact, domain_model, results)
# Domain alignment validation
self._validate_domain_alignment(dsl_artifact, domain_model, results)
# Generate suggestions
self._generate_suggestions(results)
logger.info(f"Validation completed. Valid: {results.is_completely_valid()}")
return results
def _validate_syntax(self, dsl_artifact: DSLArtifact, results: ValidationResults):
"""Validate syntax correctness of the DSL."""
grammar = dsl_artifact.grammar
implementation = dsl_artifact.implementation
# Check for balanced brackets
bracket_pairs = {'(': ')', '[': ']', '{': '}'}
stack = []
for char in implementation:
if char in bracket_pairs:
stack.append(char)
elif char in bracket_pairs.values():
if not stack:
results.syntax_valid = False
results.syntax_errors.append(f"Unmatched closing bracket: {char}")
else:
opening = stack.pop()
if bracket_pairs[opening] != char:
results.syntax_valid = False
results.syntax_errors.append(f"Mismatched brackets: {opening} and {char}")
if stack:
results.syntax_valid = False
results.syntax_errors.append(f"Unclosed brackets: {stack}")
# Check for basic Python syntax by attempting to compile
try:
compile(implementation, '<dsl_implementation>', 'exec')
except SyntaxError as e:
results.syntax_valid = False
results.syntax_errors.append(f"Python syntax error: {e}")
def _validate_semantics(self, dsl_artifact: DSLArtifact, domain_model: DomainModel, results: ValidationResults):
"""Validate semantic consistency of the DSL."""
implementation = dsl_artifact.implementation
# Check if domain entities are referenced in implementation
entity_names = [entity.name.lower() for entity in domain_model.entities]
for entity_name in entity_names:
if entity_name not in implementation.lower():
results.semantic_warnings.append(f"Domain entity '{entity_name}' not found in implementation")
# Check for common semantic issues
if 'class' not in implementation:
results.semantic_warnings.append("No class definitions found in implementation")
if 'def' not in implementation:
results.semantic_warnings.append("No method definitions found in implementation")
def _validate_domain_alignment(self, dsl_artifact: DSLArtifact, domain_model: DomainModel, results: ValidationResults):
"""Validate alignment with domain requirements."""
implementation = dsl_artifact.implementation
# Check for financial domain-specific patterns
financial_keywords = ['price', 'quantity', 'order', 'trade', 'position', 'portfolio']
found_keywords = [kw for kw in financial_keywords if kw in implementation.lower()]
if len(found_keywords) < 2:
results.domain_aligned = False
results.domain_issues.append("Insufficient financial domain terminology in implementation")
# Check if constraints are addressed
if domain_model.constraints and 'validate' not in implementation.lower():
results.domain_issues.append("Domain constraints not addressed in implementation")
def _generate_suggestions(self, results: ValidationResults):
"""Generate improvement suggestions based on validation results."""
if not results.syntax_valid:
results.suggestions.append("Fix syntax errors before proceeding with semantic validation")
if results.semantic_warnings:
results.suggestions.append("Consider adding missing domain entity references")
if not results.domain_aligned:
results.suggestions.append("Enhance domain-specific terminology and patterns")
class DSLGenerator:
"""Main DSL generation orchestrator."""
def __init__(self, llm_provider: LLMProvider, input_processor: InputProcessor,
validator: DSLValidator, template_manager: PromptTemplateManager):
self.llm_provider = llm_provider
self.input_processor = input_processor
self.validator = validator
self.template_manager = template_manager
async def generate_from_natural_language(self, description: str) -> DSLArtifact:
"""Generate DSL from natural language description."""
logger.info("Starting DSL generation from natural language")
# Process input
domain_model = self.input_processor.process_natural_language(description)
# Generate DSL
dsl_artifact = await self._generate_dsl_artifact(domain_model)
# Validate and refine
dsl_artifact = await self._validate_and_refine(dsl_artifact, domain_model)
return dsl_artifact
async def generate_from_ddd_specification(self, ddd_spec: Dict[str, Any]) -> DSLArtifact:
"""Generate DSL from DDD specification."""
logger.info("Starting DSL generation from DDD specification")
# Process input
domain_model = self.input_processor.process_ddd_specification(ddd_spec)
# Generate DSL
dsl_artifact = await self._generate_dsl_artifact(domain_model)
# Validate and refine
dsl_artifact = await self._validate_and_refine(dsl_artifact, domain_model)
return dsl_artifact
async def _generate_dsl_artifact(self, domain_model: DomainModel) -> DSLArtifact:
"""Generate DSL artifact from domain model."""
prompt = self.template_manager.format_template(
'dsl_generation',
domain_model=json.dumps(domain_model.to_dict(), indent=2)
)
try:
response = await self.llm_provider.generate(
prompt=prompt,
max_tokens=2048,
temperature=0.3
)
# Parse response to extract grammar and implementation
grammar, implementation = self._parse_llm_response(response)
return DSLArtifact(
grammar=grammar,
implementation=implementation,
metadata={
'domain_model': domain_model.to_dict(),
'generation_method': 'llm_based',
'llm_provider': type(self.llm_provider).__name__
}
)
except Exception as e:
logger.error(f"DSL generation failed: {e}")
raise DSLGenerationError(f"Failed to generate DSL: {e}")
def _parse_llm_response(self, response: str) -> tuple[str, str]:
"""Parse LLM response to extract grammar and implementation."""
# Simple parsing logic - in production, use more sophisticated parsing
lines = response.split('\n')
grammar_start = -1
implementation_start = -1
for i, line in enumerate(lines):
if 'grammar' in line.lower() or 'ebnf' in line.lower():
grammar_start = i
elif 'implementation' in line.lower() or 'python' in line.lower():
implementation_start = i
break
if grammar_start >= 0 and implementation_start > grammar_start:
grammar = '\n'.join(lines[grammar_start:implementation_start])
implementation = '\n'.join(lines[implementation_start:])
else:
# Fallback: treat entire response as implementation
grammar = "# Grammar not clearly separated"
implementation = response
return grammar.strip(), implementation.strip()
async def _validate_and_refine(self, dsl_artifact: DSLArtifact, domain_model: DomainModel) -> DSLArtifact:
"""Validate DSL and refine if necessary."""
validation_results = self.validator.validate_dsl(dsl_artifact, domain_model)
dsl_artifact.validation_results = validation_results
if not validation_results.is_completely_valid():
logger.info("DSL validation failed, attempting refinement")
# Attempt refinement
refined_artifact = await self._refine_dsl(dsl_artifact, validation_results, domain_model)
if refined_artifact:
return refined_artifact
return dsl_artifact
async def _refine_dsl(self, dsl_artifact: DSLArtifact, validation_results: ValidationResults,
domain_model: DomainModel) -> Optional[DSLArtifact]:
"""Refine DSL based on validation issues."""
issues_summary = self._summarize_validation_issues(validation_results)
prompt = self.template_manager.format_template(
'refinement',
original_dsl=dsl_artifact.implementation,
validation_issues=issues_summary
)
try:
response = await self.llm_provider.generate(
prompt=prompt,
max_tokens=2048,
temperature=0.2
)
grammar, implementation = self._parse_llm_response(response)
refined_artifact = DSLArtifact(
grammar=grammar,
implementation=implementation,
metadata={
**dsl_artifact.metadata,
'refinement_attempt': True,
'original_issues': issues_summary
}
)
# Validate refined version
refined_validation = self.validator.validate_dsl(refined_artifact, domain_model)
refined_artifact.validation_results = refined_validation
return refined_artifact
except Exception as e:
logger.error(f"DSL refinement failed: {e}")
return None
def _summarize_validation_issues(self, validation_results: ValidationResults) -> str:
"""Create a summary of validation issues for refinement prompt."""
issues = []
if validation_results.syntax_errors:
issues.append(f"Syntax errors: {'; '.join(validation_results.syntax_errors)}")
if validation_results.semantic_warnings:
issues.append(f"Semantic warnings: {'; '.join(validation_results.semantic_warnings)}")
if validation_results.domain_issues:
issues.append(f"Domain issues: {'; '.join(validation_results.domain_issues)}")
return '\n'.join(issues)
class TradingDSLSystem:
"""Complete trading DSL generation system."""
def __init__(self, llm_provider: LLMProvider):
self.llm_provider = llm_provider
self.input_processor = InputProcessor()
self.validator = DSLValidator()
self.template_manager = PromptTemplateManager()
self.generator = DSLGenerator(
llm_provider=llm_provider,
input_processor=self.input_processor,
validator=self.validator,
template_manager=self.template_manager
)
async def generate_trading_dsl(self, input_spec: Union[str, Dict[str, Any]]) -> DSLArtifact:
"""Generate trading DSL from input specification."""
if isinstance(input_spec, str):
return await self.generator.generate_from_natural_language(input_spec)
else:
return await self.generator.generate_from_ddd_specification(input_spec)
def save_dsl_artifact(self, artifact: DSLArtifact, filepath: str):
"""Save DSL artifact to file."""
with open(filepath, 'w') as f:
json.dump(artifact.to_dict(), f, indent=2, default=str)
logger.info(f"DSL artifact saved to {filepath}")
def load_dsl_artifact(self, filepath: str) -> DSLArtifact:
"""Load DSL artifact from file."""
with open(filepath, 'r') as f:
data = json.load(f)
artifact = DSLArtifact(
grammar=data['grammar'],
implementation=data['implementation'],
metadata=data['metadata']
)
if data.get('validation_results'):
validation_data = data['validation_results']
artifact.validation_results = ValidationResults(
syntax_valid=validation_data['syntax_valid'],
semantic_valid=validation_data['semantic_valid'],
domain_aligned=validation_data['domain_aligned'],
syntax_errors=validation_data['syntax_errors'],
semantic_warnings=validation_data['semantic_warnings'],
domain_issues=validation_data['domain_issues'],
suggestions=validation_data['suggestions']
)
logger.info(f"DSL artifact loaded from {filepath}")
return artifact
# Example usage and demonstration
async def main():
"""Demonstrate the complete DSL generation system."""
# Initialize with a mock local LLM provider for demonstration
# In practice, use actual model or remote provider
class MockLLMProvider(LLMProvider):
async def generate(self, prompt: str, **kwargs) -> str:
# Mock response for demonstration
return """
Grammar Definition (EBNF):
trading_rule ::= "RULE" identifier ":" condition "THEN" action
condition ::= expression ("AND" | "OR") expression | expression
expression ::= identifier operator value
action ::= "BUY" | "SELL" | "HOLD" | "ALERT"
Python Implementation:
import re
from typing import Dict, List, Any
from dataclasses import dataclass
from enum import Enum
class ActionType(Enum):
BUY = "BUY"
SELL = "SELL"
HOLD = "HOLD"
ALERT = "ALERT"
@dataclass
class TradingRule:
name: str
condition: str
action: ActionType
parameters: Dict[str, Any] = None
def evaluate(self, market_data: Dict[str, float]) -> bool:
# Simple condition evaluation
return eval(self.condition, {"__builtins__": {}}, market_data)
class TradingDSLParser:
def __init__(self):
self.rules = []
def parse_rule(self, rule_text: str) -> TradingRule:
# Parse DSL rule text into TradingRule object
pattern = r'RULE\s+(\w+):\s*(.+?)\s+THEN\s+(\w+)'
match = re.match(pattern, rule_text, re.IGNORECASE)
if not match:
raise ValueError(f"Invalid rule syntax: {rule_text}")
name, condition, action = match.groups()
return TradingRule(
name=name,
condition=condition,
action=ActionType(action.upper())
)
def add_rule(self, rule_text: str):
rule = self.parse_rule(rule_text)
self.rules.append(rule)
def evaluate_rules(self, market_data: Dict[str, float]) -> List[TradingRule]:
triggered_rules = []
for rule in self.rules:
try:
if rule.evaluate(market_data):
triggered_rules.append(rule)
except Exception as e:
print(f"Error evaluating rule {rule.name}: {e}")
return triggered_rules
# Example usage:
parser = TradingDSLParser()
parser.add_rule("RULE momentum_buy: price > sma_20 AND volume > avg_volume THEN BUY")
parser.add_rule("RULE stop_loss: price < entry_price * 0.95 THEN SELL")
market_data = {
'price': 150.0,
'sma_20': 145.0,
'volume': 1000000,
'avg_volume': 800000,
'entry_price': 148.0
}
triggered = parser.evaluate_rules(market_data)
for rule in triggered:
print(f"Rule {rule.name} triggered: {rule.action}")
"""
def is_available(self) -> bool:
return True
# Initialize system
llm_provider = MockLLMProvider()
trading_system = TradingDSLSystem(llm_provider)
# Example 1: Generate DSL from natural language
print("=== Generating DSL from Natural Language ===")
natural_language_spec = """
Create a trading system that can handle buy and sell orders based on technical indicators.
Orders should have symbol, quantity, and price. The system should support momentum trading
rules that trigger when price crosses above moving averages with high volume.
Include stop-loss rules that sell positions when price drops below a threshold.
"""
try:
dsl_artifact = await trading_system.generate_trading_dsl(natural_language_spec)
print(f"Generated DSL Grammar:\n{dsl_artifact.grammar}\n")
print(f"Generated Implementation:\n{dsl_artifact.implementation}\n")
print(f"Validation Results: {dsl_artifact.validation_results.to_dict()}\n")
# Save artifact
trading_system.save_dsl_artifact(dsl_artifact, "trading_dsl_nl.json")
except Exception as e:
print(f"Error generating DSL from natural language: {e}")
# Example 2: Generate DSL from DDD specification
print("=== Generating DSL from DDD Specification ===")
ddd_spec = {
"bounded_contexts": ["Trading", "RiskManagement", "MarketData"],
"entities": [
{
"name": "Order",
"attributes": {
"symbol": "string",
"quantity": "number",
"price": "number",
"side": "string",
"timestamp": "datetime"
},
"constraints": ["quantity > 0", "price > 0"],
"relationships": ["belongs_to Portfolio"]
},
{
"name": "Position",
"attributes": {
"symbol": "string",
"quantity": "number",
"average_price": "number",
"unrealized_pnl": "number"
},
"constraints": ["quantity != 0"],
"relationships": ["part_of Portfolio"]
},
{
"name": "TradingRule",
"attributes": {
"name": "string",
"condition": "string",
"action": "string",
"active": "boolean"
},
"constraints": ["name must be unique"],
"relationships": ["applies_to Strategy"]
}
],
"domain_events": [
"OrderPlaced",
"OrderFilled",
"PositionOpened",
"PositionClosed",
"RuleTriggered"
],
"relationships": [
"Portfolio contains Orders",
"Portfolio contains Positions",
"Strategy contains TradingRules"
],
"constraints": [
"Total position value cannot exceed portfolio limit",
"Risk per trade cannot exceed 2% of portfolio",
"Maximum 10 active rules per strategy"
]
}
try:
dsl_artifact = await trading_system.generate_trading_dsl(ddd_spec)
print(f"Generated DSL Grammar:\n{dsl_artifact.grammar}\n")
print(f"Generated Implementation:\n{dsl_artifact.implementation}\n")
print(f"Validation Results: {dsl_artifact.validation_results.to_dict()}\n")
# Save artifact
trading_system.save_dsl_artifact(dsl_artifact, "trading_dsl_ddd.json")
except Exception as e:
print(f"Error generating DSL from DDD specification: {e}")
print("=== DSL Generation Complete ===")
if __name__ == "__main__":
asyncio.run(main())
This complete running example demonstrates a production-ready LLM-based DSL generation system specifically designed for financial trading domains. The implementation includes comprehensive error handling, validation frameworks, support for both local and remote LLM providers, and demonstrates the conversion of both natural language descriptions and formal DDD specifications into executable domain-specific languages.
The system architecture follows clean code principles with clear separation of concerns, comprehensive logging, and robust error handling. The example includes realistic financial domain modeling and generates practical DSL implementations that could be used in actual trading systems with appropriate extensions and refinements.
No comments:
Post a Comment