Design Principles¶
This document outlines the core design principles and architectural philosophy that guide FinWiz's development and evolution.
Core Philosophy¶
AI-Powered, Human-Guided¶
FinWiz leverages AI for analysis while maintaining human oversight and control:
- AI for Analysis: Use CrewAI agents for complex financial analysis requiring reasoning
- Python for Logic: Use deterministic Python code for calculations and data processing
- Human for Decisions: Provide recommendations, not automated trading decisions
- Transparency: Always show reasoning and data sources behind recommendations
Quality Over Speed¶
Prioritize analysis quality and accuracy over execution speed:
- Thorough Analysis: Comprehensive evaluation using multiple data sources
- Validation First: Strict data validation at all system boundaries
- Fail Fast: Reject invalid data rather than attempting to fix it
- No Hallucinations: Never generate fake data to fill gaps
Modular and Extensible¶
Design for flexibility and future growth:
- Crew-Based Architecture: Specialized crews for different asset classes
- Tool Composition: Reusable tools that can be combined in different ways
- Schema-Driven: Pydantic models define clear contracts between components
- Plugin Architecture: Easy to add new analysis capabilities
Architectural Principles¶
1. Separation of Concerns¶
Each component has a single, well-defined responsibility:
Python
# ✅ Good: Single responsibility
class TickerValidationTool:
"""Validates ticker symbols only."""
def validate(self, ticker: str) -> ValidationResult:
pass
class RiskAssessmentTool:
"""Calculates risk metrics only."""
def assess_risk(self, data: dict) -> RiskAssessment:
pass
# ❌ Bad: Multiple responsibilities
class AnalysisTool:
"""Does everything - validation, analysis, risk, reporting."""
def do_everything(self, ticker: str) -> CompleteReport:
pass
2. Explicit Dependencies¶
Make dependencies clear and manageable:
Python
# ✅ Good: Explicit dependencies
class StockAnalyzer:
def __init__(self,
validator: TickerValidationTool,
data_source: YahooFinanceTool,
risk_assessor: RiskAssessmentTool):
self.validator = validator
self.data_source = data_source
self.risk_assessor = risk_assessor
# ❌ Bad: Hidden dependencies
class StockAnalyzer:
def analyze(self, ticker: str):
# Hidden dependency on global state
data = global_data_source.get_data(ticker)
3. Immutable Data Flow¶
Data flows through the system without mutation:
Python
# ✅ Good: Immutable data flow
def analyze_stock(ticker: str) -> StockAnalysis:
raw_data = fetch_data(ticker)
validated_data = validate_data(raw_data)
analysis = perform_analysis(validated_data)
return analysis
# ❌ Bad: Mutable shared state
global_data = {}
def analyze_stock(ticker: str):
global_data[ticker] = fetch_data(ticker)
modify_global_data(ticker) # Mutation
4. Contract-First Design¶
Define interfaces before implementation:
Python
# ✅ Good: Clear contract
class AnalysisTool(Protocol):
def analyze(self, ticker: str) -> AnalysisResult:
"""Analyze a ticker and return structured results."""
...
# Implementation follows contract
class StockAnalysisTool:
def analyze(self, ticker: str) -> AnalysisResult:
# Implementation details
pass
Data Principles¶
1. Schema-Driven Validation¶
All data structures use Pydantic models with strict validation:
Python
class StockAnalysis(BaseModel):
ticker: str = Field(..., pattern=r'^[A-Z]{1,5}$')
grade: str = Field(..., pattern=r'^(A\+|A|B|C|D|F)$')
composite_score: float = Field(..., ge=0.0, le=1.0)
model_config = ConfigDict(
extra='forbid', # Reject unknown fields
str_strip_whitespace=True, # Clean input data
validate_assignment=True # Validate on assignment
)
2. Source Attribution¶
Every data point must be traceable to its source:
Python
class DataPoint(BaseModel):
value: float
source: str = Field(..., description="Data source name")
source_url: Optional[str] = None
retrieved_at: datetime
confidence: float = Field(..., ge=0.0, le=1.0)
3. Freshness Awareness¶
Track and communicate data age:
Python
class AnalysisResult(BaseModel):
# ... other fields ...
data_freshness: Dict[str, datetime]
stale_data_warning: bool = False
@model_validator(mode='after')
def check_data_freshness(self) -> 'AnalysisResult':
now = datetime.now()
for source, timestamp in self.data_freshness.items():
age_hours = (now - timestamp).total_seconds() / 3600
if age_hours > 24: # Configurable threshold
self.stale_data_warning = True
return self
4. Graceful Degradation¶
Handle missing data transparently:
Python
# ✅ Good: Explicit handling of missing data
def calculate_score(fundamental: Optional[float],
technical: Optional[float]) -> ScoreResult:
if fundamental is None and technical is None:
return ScoreResult(
score=None,
confidence=0.0,
warning="Insufficient data for scoring"
)
# Calculate with available data
available_scores = [s for s in [fundamental, technical] if s is not None]
return ScoreResult(
score=sum(available_scores) / len(available_scores),
confidence=len(available_scores) / 2.0,
warning=None if len(available_scores) == 2 else "Partial data used"
)
AI Integration Principles¶
1. AI Minimalism¶
Use AI only where human-like reasoning is required:
Python
# ✅ Good: AI for reasoning tasks
class InvestmentAnalyst(Agent):
"""Uses AI to interpret complex financial data and market conditions."""
def analyze_investment_thesis(self, data: dict) -> InvestmentThesis:
# AI reasoning about qualitative factors
pass
# ✅ Good: Python for deterministic tasks
def calculate_sharpe_ratio(returns: List[float], risk_free_rate: float) -> float:
"""Deterministic calculation - no AI needed."""
excess_returns = [r - risk_free_rate for r in returns]
return statistics.mean(excess_returns) / statistics.stdev(excess_returns)
2. Reasoning Transparency¶
Make AI reasoning visible and auditable:
Python
class AnalysisResult(BaseModel):
recommendation: str
reasoning_steps: List[str] # Show how AI reached conclusion
data_sources: List[str] # Show what data was used
confidence_factors: Dict[str, float] # Show confidence breakdown
3. Bounded AI Scope¶
Limit AI agent scope to prevent hallucinations:
Python
# ✅ Good: Bounded scope
class RiskAssessmentAgent(Agent):
"""Focused on risk assessment only."""
tools = [RiskCalculationTool, VolatilityTool] # Limited tool set
def assess_risk(self, ticker: str) -> RiskAssessment:
# Focused task with clear boundaries
pass
# ❌ Bad: Unbounded scope
class GeneralAnalysisAgent(Agent):
"""Does everything - prone to hallucinations."""
tools = [AllTools] # Too many tools, unclear boundaries
Performance Principles¶
1. Optimize for Common Cases¶
Design for typical usage patterns:
Python
# ✅ Good: Optimized for common case (single ticker analysis)
def analyze_ticker(ticker: str) -> AnalysisResult:
# Fast path for single ticker
pass
def analyze_portfolio(tickers: List[str]) -> List[AnalysisResult]:
# Batch processing for multiple tickers
return [analyze_ticker(t) for t in tickers]
2. Intelligent Caching¶
Cache expensive operations with appropriate TTL:
Python
@cache(ttl=3600) # 1 hour cache
def fetch_market_data(ticker: str) -> MarketData:
# Expensive API call
pass
@cache(ttl=86400) # 24 hour cache
def fetch_fundamental_data(ticker: str) -> FundamentalData:
# Very expensive analysis
pass
3. Async Where Appropriate¶
Use async for I/O-bound operations:
Python
# ✅ Good: Async for I/O
async def fetch_multiple_tickers(tickers: List[str]) -> Dict[str, MarketData]:
tasks = [fetch_market_data(ticker) for ticker in tickers]
results = await asyncio.gather(*tasks)
return dict(zip(tickers, results))
# ✅ Good: Sync for CPU-bound
def calculate_technical_indicators(prices: List[float]) -> TechnicalIndicators:
# CPU-bound calculation - no need for async
pass
Error Handling Principles¶
1. Fail Fast and Loud¶
Detect errors early and communicate clearly:
Python
# ✅ Good: Fail fast with clear error
def analyze_ticker(ticker: str) -> AnalysisResult:
if not re.match(r'^[A-Z]{1,5}$', ticker):
raise ValueError(f"Invalid ticker format: {ticker}. Must be 1-5 uppercase letters.")
# Continue with valid ticker
pass
# ❌ Bad: Silent failure or unclear error
def analyze_ticker(ticker: str) -> Optional[AnalysisResult]:
if not is_valid_ticker(ticker):
return None # Silent failure - user doesn't know why
2. Contextual Error Messages¶
Provide actionable error information:
Python
class AnalysisError(Exception):
def __init__(self, ticker: str, reason: str, remediation: str):
self.ticker = ticker
self.reason = reason
self.remediation = remediation
super().__init__(f"Analysis failed for {ticker}: {reason}. {remediation}")
# Usage
raise AnalysisError(
ticker="INVALID",
reason="Ticker not found in market data",
remediation="Check ticker spelling or try a different symbol"
)
3. Graceful Recovery¶
Provide fallback options when possible:
Python
def get_market_data(ticker: str) -> MarketData:
try:
return primary_data_source.get_data(ticker)
except APIError as e:
logger.warning(f"Primary source failed: {e}")
try:
data = fallback_data_source.get_data(ticker)
data.source = "fallback"
data.confidence *= 0.8 # Reduce confidence for fallback data
return data
except APIError:
raise DataUnavailableError(f"No data available for {ticker}")
Security Principles¶
1. Input Validation¶
Validate all external inputs:
Python
class TickerInput(BaseModel):
ticker: str = Field(..., pattern=r'^[A-Z]{1,5}$')
@field_validator('ticker')
@classmethod
def validate_ticker(cls, v: str) -> str:
# Additional validation logic
if v in BLACKLISTED_TICKERS:
raise ValueError(f"Ticker {v} is not supported")
return v.upper()
2. Secure API Key Management¶
Never log or expose API keys:
Python
# ✅ Good: Secure key handling
class APIClient:
def __init__(self):
self.api_key = os.getenv("API_KEY")
if not self.api_key:
raise ConfigurationError("API_KEY environment variable not set")
def make_request(self, url: str) -> dict:
headers = {"Authorization": f"Bearer {self.api_key}"}
# Never log headers or api_key
logger.info(f"Making request to {url}") # Safe to log URL
return requests.get(url, headers=headers).json()
3. Data Privacy¶
Never log sensitive financial information:
Python
# ✅ Good: Privacy-aware logging
def analyze_portfolio(holdings: List[Holding]) -> PortfolioAnalysis:
logger.info(f"Analyzing portfolio with {len(holdings)} holdings") # Safe
# Never log: actual holdings, values, or personal information
for holding in holdings:
# Process holding
logger.debug(f"Processing {holding.ticker}") # Ticker is public info
# Never log: holding.value, holding.quantity, etc.
Testing Principles¶
1. Test Behavior, Not Implementation¶
Focus on what the code does, not how it does it:
Python
# ✅ Good: Test behavior
def test_should_return_buy_recommendation_for_strong_stock():
# Arrange
strong_stock_data = create_strong_stock_data()
# Act
result = analyze_stock(strong_stock_data)
# Assert
assert result.recommendation == "BUY"
assert result.confidence > 0.8
# ❌ Bad: Test implementation details
def test_should_call_risk_calculator_with_correct_parameters():
# Testing internal method calls instead of behavior
pass
2. Mock External Dependencies¶
Mock all external systems and APIs:
Python
def test_should_handle_api_failure_gracefully(mocker):
# Mock external API to simulate failure
mock_api = mocker.patch('finwiz.tools.yahoo_finance_tool.get_data')
mock_api.side_effect = APIError("Service unavailable")
# Test graceful handling
result = analyze_stock("AAPL")
assert result.error is not None
assert "Service unavailable" in result.error
3. Test Edge Cases¶
Include tests for boundary conditions and error cases:
Python
def test_should_handle_invalid_ticker_format():
with pytest.raises(ValueError, match="Invalid ticker format"):
analyze_stock("invalid_ticker")
def test_should_handle_missing_data():
result = analyze_stock_with_missing_data("AAPL")
assert result.confidence < 0.5
assert "missing data" in result.warnings
Documentation Principles¶
1. Code as Documentation¶
Write self-documenting code:
Python
# ✅ Good: Self-documenting
def calculate_risk_adjusted_return(returns: List[float],
risk_free_rate: float) -> float:
"""Calculate Sharpe ratio (risk-adjusted return)."""
excess_returns = [r - risk_free_rate for r in returns]
return statistics.mean(excess_returns) / statistics.stdev(excess_returns)
# ❌ Bad: Unclear purpose
def calc(data: List[float], rate: float) -> float:
# What does this calculate?
pass
2. Document Decisions, Not Code¶
Explain why, not what:
Python
class PortfolioAnalyzer:
def __init__(self, max_position_size: float = 0.1):
# Why: Limit single position to 10% to manage concentration risk
# This follows modern portfolio theory recommendations
self.max_position_size = max_position_size
3. Keep Documentation Current¶
Documentation should evolve with code:
Python
# Update docstrings when behavior changes
def analyze_stock(ticker: str, deep_analysis: bool = False) -> AnalysisResult:
"""
Analyze a stock ticker.
Args:
ticker: Stock symbol (1-5 uppercase letters)
deep_analysis: Enable comprehensive analysis (added in v2.0)
Returns:
AnalysisResult with recommendation and confidence
Raises:
ValueError: If ticker format is invalid
APIError: If market data is unavailable
"""
Evolution Principles¶
1. Backward Compatibility¶
Maintain compatibility when possible:
Python
# ✅ Good: Backward compatible change
def analyze_stock(ticker: str,
deep_analysis: bool = False, # New optional parameter
**kwargs) -> AnalysisResult:
# Old calls still work, new functionality available
pass
# ❌ Bad: Breaking change
def analyze_stock(ticker: str,
analysis_type: str) -> AnalysisResult: # Required new parameter
# Breaks existing code
pass
2. Gradual Migration¶
Provide migration paths for breaking changes:
Python
# Phase 1: Deprecate old method
@deprecated("Use analyze_stock_v2() instead. Will be removed in v3.0")
def analyze_stock(ticker: str) -> AnalysisResult:
return analyze_stock_v2(ticker, legacy_mode=True)
# Phase 2: New method with improved interface
def analyze_stock_v2(ticker: str, options: AnalysisOptions) -> AnalysisResult:
pass
# Phase 3: Remove deprecated method in next major version
3. Feature Flags¶
Use feature flags for gradual rollouts:
Python
def analyze_stock(ticker: str) -> AnalysisResult:
if feature_flags.is_enabled("enhanced_analysis"):
return enhanced_analyze_stock(ticker)
else:
return legacy_analyze_stock(ticker)
Conclusion¶
These design principles guide FinWiz's architecture and development:
- Quality First: Prioritize accuracy and reliability over speed
- Modular Design: Build composable, reusable components
- AI Minimalism: Use AI where it adds value, Python elsewhere
- Fail Fast: Detect and communicate errors early
- Transparency: Make reasoning and data sources visible
- Security: Protect sensitive data and validate inputs
- Evolution: Design for change while maintaining stability
By following these principles, FinWiz maintains high code quality, reliability, and extensibility while providing accurate financial analysis.
Version: 2.0 Last Updated: 2025-10-26