main.py•34.3 kB
# main.py - Complete Enhanced Trading System with All Features
import pandas as pd
import numpy as np
import warnings
import json
import pickle
import argparse
from datetime import datetime, timedelta
from pathlib import Path
from typing import List, Dict, Any, Optional, Set, Tuple
import matplotlib.pyplot as plt
import seaborn as sns
warnings.filterwarnings('ignore')
from models.arima_model import get_arima_forecast
from models.hybrid_model import train_xgboost_on_residuals
class SmartAnalysisCache:
"""Cache analysis results to avoid duplicate processing."""
def __init__(self, cache_dir: str = "cache", cache_hours: int = 6):
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(exist_ok=True)
self.cache_hours = cache_hours
self.session_cache = {}
def get_cache_key(self, ticker: str) -> str:
"""Generate cache key for a ticker."""
date_str = datetime.now().strftime("%Y%m%d_%H")
return f"{ticker}_{date_str}"
def is_cache_valid(self, cache_file: Path) -> bool:
"""Check if cache file is still valid."""
if not cache_file.exists():
return False
cache_time = datetime.fromtimestamp(cache_file.stat().st_mtime)
cutoff_time = datetime.now() - timedelta(hours=self.cache_hours)
return cache_time > cutoff_time
def get_cached_result(self, ticker: str) -> Optional[Dict[str, Any]]:
"""Get cached analysis result for a ticker."""
if ticker in self.session_cache:
return self.session_cache[ticker]
cache_key = self.get_cache_key(ticker)
cache_file = self.cache_dir / f"{cache_key}.pkl"
if self.is_cache_valid(cache_file):
try:
with open(cache_file, 'rb') as f:
result = pickle.load(f)
self.session_cache[ticker] = result
return result
except Exception:
pass
return None
def save_result(self, ticker: str, result: Dict[str, Any]):
"""Save analysis result to cache."""
self.session_cache[ticker] = result
cache_key = self.get_cache_key(ticker)
cache_file = self.cache_dir / f"{cache_key}.pkl"
try:
with open(cache_file, 'wb') as f:
pickle.dump(result, f)
except Exception as e:
print(f"⚠️ Cache save failed for {ticker}: {e}")
class PortfolioComparator:
"""Compare multiple portfolios with various metrics."""
def __init__(self):
self.comparison_metrics = {}
def calculate_portfolio_metrics(self, portfolio_results: List[Dict[str, Any]], portfolio_name: str) -> Dict[str, Any]:
"""Calculate comprehensive portfolio metrics."""
if not portfolio_results:
return {"portfolio_name": portfolio_name, "error": "No valid results"}
expected_changes = [r['expected_change'] for r in portfolio_results]
confidences = [r['confidence'] for r in portfolio_results]
volatilities = [r['volatility'] for r in portfolio_results]
signals = [r['signal'] for r in portfolio_results]
buy_signals = signals.count('BUY')
sell_signals = signals.count('SELL')
hold_signals = signals.count('HOLD')
portfolio_return = np.mean(expected_changes)
portfolio_volatility = np.mean(volatilities)
portfolio_confidence = np.mean(confidences)
max_gain = max(expected_changes)
max_loss = min(expected_changes)
positive_stocks = len([x for x in expected_changes if x > 0])
negative_stocks = len([x for x in expected_changes if x < 0])
risk_adjusted_return = portfolio_return / portfolio_volatility if portfolio_volatility > 0 else 0
high_conf_signals = [r for r in portfolio_results if r['confidence'] > 0.7 and r['signal'] != 'HOLD']
volatility_spread = np.std(volatilities) if len(volatilities) > 1 else 0
diversification_score = max(0, 1 - volatility_spread)
return {
"portfolio_name": portfolio_name,
"num_stocks": len(portfolio_results),
"expected_return": portfolio_return,
"max_gain_stock": max_gain,
"max_loss_stock": max_loss,
"positive_stocks": positive_stocks,
"negative_stocks": negative_stocks,
"positive_ratio": positive_stocks / len(portfolio_results),
"portfolio_volatility": portfolio_volatility,
"risk_adjusted_return": risk_adjusted_return,
"diversification_score": diversification_score,
"avg_confidence": portfolio_confidence,
"high_confidence_opportunities": len(high_conf_signals),
"buy_signals": buy_signals,
"sell_signals": sell_signals,
"hold_signals": hold_signals,
"action_ratio": (buy_signals + sell_signals) / len(portfolio_results),
"quality_score": self._calculate_quality_score(
portfolio_return, portfolio_volatility, portfolio_confidence,
diversification_score, len(high_conf_signals)
),
"stocks": portfolio_results
}
def _calculate_quality_score(self, portfolio_return: float, volatility: float,
confidence: float, diversification: float, high_conf_ops: int) -> float:
"""Calculate composite portfolio quality score (0-100)."""
return_score = min(1, max(0, (portfolio_return + 0.05) / 0.1))
volatility_score = max(0, 1 - volatility / 0.1)
confidence_score = confidence
diversification_score = diversification
opportunity_score = min(1, high_conf_ops / 3)
quality_score = (
return_score * 0.3 +
volatility_score * 0.2 +
confidence_score * 0.2 +
diversification_score * 0.15 +
opportunity_score * 0.15
) * 100
return round(quality_score, 1)
def compare_portfolios(self, portfolio_metrics: Dict[str, Dict[str, Any]]) -> Dict[str, Any]:
"""Compare multiple portfolios and provide rankings."""
if not portfolio_metrics:
return {"error": "No portfolios to compare"}
comparison_data = []
for portfolio_name, metrics in portfolio_metrics.items():
if 'error' not in metrics:
comparison_data.append({
'Portfolio': portfolio_name,
'Expected Return': f"{metrics['expected_return']:+.2%}",
'Volatility': f"{metrics['portfolio_volatility']:.2%}",
'Quality Score': metrics['quality_score'],
'Risk-Adj Return': f"{metrics['risk_adjusted_return']:.2f}",
'Confidence': f"{metrics['avg_confidence']:.1%}",
'Action Ratio': f"{metrics['action_ratio']:.1%}",
'Buy Signals': metrics['buy_signals'],
'Sell Signals': metrics['sell_signals'],
'High Conf Ops': metrics['high_confidence_opportunities'],
'Stocks': metrics['num_stocks']
})
if not comparison_data:
return {"error": "No valid portfolio data for comparison"}
df = pd.DataFrame(comparison_data)
rankings = {
'highest_return': self._get_top_portfolios(portfolio_metrics, 'expected_return', 3),
'lowest_risk': self._get_top_portfolios(portfolio_metrics, 'portfolio_volatility', 3, ascending=True),
'highest_quality': self._get_top_portfolios(portfolio_metrics, 'quality_score', 3),
'most_confident': self._get_top_portfolios(portfolio_metrics, 'avg_confidence', 3),
'most_opportunities': self._get_top_portfolios(portfolio_metrics, 'high_confidence_opportunities', 3),
'best_risk_adjusted': self._get_top_portfolios(portfolio_metrics, 'risk_adjusted_return', 3)
}
recommendations = self._generate_recommendations(portfolio_metrics)
return {
'comparison_table': df,
'rankings': rankings,
'recommendations': recommendations,
'summary_stats': self._get_summary_stats(portfolio_metrics)
}
def _get_top_portfolios(self, portfolio_metrics: Dict, metric: str, top_n: int, ascending: bool = False) -> List[Tuple[str, float]]:
"""Get top N portfolios for a specific metric."""
valid_portfolios = [(name, metrics[metric]) for name, metrics in portfolio_metrics.items()
if 'error' not in metrics and metric in metrics]
sorted_portfolios = sorted(valid_portfolios, key=lambda x: x[1], reverse=not ascending)
return sorted_portfolios[:top_n]
def _generate_recommendations(self, portfolio_metrics: Dict) -> Dict[str, str]:
"""Generate portfolio recommendations based on different investor profiles."""
conservative_pick = max(
[(name, metrics) for name, metrics in portfolio_metrics.items() if 'error' not in metrics],
key=lambda x: x[1]['diversification_score'] - x[1]['portfolio_volatility'],
default=(None, {})
)
aggressive_pick = max(
[(name, metrics) for name, metrics in portfolio_metrics.items() if 'error' not in metrics],
key=lambda x: x[1]['expected_return'],
default=(None, {})
)
balanced_pick = max(
[(name, metrics) for name, metrics in portfolio_metrics.items() if 'error' not in metrics],
key=lambda x: x[1]['quality_score'],
default=(None, {})
)
return {
'conservative': f"{conservative_pick[0]} (Low risk, diversified)" if conservative_pick[0] else "No suitable portfolio",
'aggressive': f"{aggressive_pick[0]} (Highest expected return)" if aggressive_pick[0] else "No suitable portfolio",
'balanced': f"{balanced_pick[0]} (Best overall quality)" if balanced_pick[0] else "No suitable portfolio"
}
def _get_summary_stats(self, portfolio_metrics: Dict) -> Dict[str, Any]:
"""Get summary statistics across all portfolios."""
valid_metrics = [metrics for metrics in portfolio_metrics.values() if 'error' not in metrics]
if not valid_metrics:
return {"error": "No valid portfolio metrics"}
returns = [m['expected_return'] for m in valid_metrics]
volatilities = [m['portfolio_volatility'] for m in valid_metrics]
quality_scores = [m['quality_score'] for m in valid_metrics]
return {
'total_portfolios': len(valid_metrics),
'avg_return_across_portfolios': np.mean(returns),
'avg_volatility_across_portfolios': np.mean(volatilities),
'avg_quality_score': np.mean(quality_scores),
'return_range': f"{min(returns):+.2%} to {max(returns):+.2%}",
'best_quality_score': max(quality_scores),
'portfolio_spread': np.std(returns)
}
class SmartTradingSystem:
"""Enhanced trading system with smart caching, portfolio comparison, and signal filtering."""
def __init__(self, config_path: str = "config/trading_config.json"):
self.config = self._load_config(config_path)
self.cache = SmartAnalysisCache()
self.comparator = PortfolioComparator()
self.analyzed_tickers = set()
# Trading parameters
trading_params = self.config.get("trading_parameters", {})
self.buy_threshold = trading_params.get('buy_threshold', 0.02)
self.sell_threshold = trading_params.get('sell_threshold', -0.02)
self.confidence_threshold = trading_params.get('confidence_threshold', 0.6)
def _load_config(self, config_path: str) -> Dict[str, Any]:
"""Load configuration from file."""
try:
with open(config_path, 'r') as f:
return json.load(f)
except Exception as e:
print(f"❌ Error loading config: {e}")
return {"tickers": {"default": ["AAPL", "MSFT"]}}
def get_all_unique_stocks(self) -> List[str]:
"""Get all unique stocks from all portfolios."""
all_stocks = set()
for portfolio_name, tickers in self.config.get("tickers", {}).items():
if isinstance(tickers, list):
all_stocks.update(tickers)
# Remove any metadata entries (strings starting with //)
all_stocks = {stock for stock in all_stocks if not str(stock).startswith('//')}
return sorted(list(all_stocks))
def analyze_stock_cached(self, ticker: str) -> Optional[Dict[str, Any]]:
"""Analyze stock with smart caching."""
# Check session cache first
if ticker in self.analyzed_tickers:
cached_result = self.cache.session_cache.get(ticker)
if cached_result:
print(f"🔄 {ticker}: Using session cache")
return cached_result
# Check disk cache
cached_result = self.cache.get_cached_result(ticker)
if cached_result:
print(f"📋 {ticker}: Using cached result")
self.analyzed_tickers.add(ticker)
return cached_result
# Run fresh analysis
print(f"🔬 {ticker}: Running fresh analysis...")
try:
# ARIMA analysis
arima_forecast, residuals, df = get_arima_forecast(ticker)
# XGBoost analysis
xgb_model, latest_features = train_xgboost_on_residuals(residuals, df)
# Calculate results
residual_correction = float(xgb_model.predict(latest_features)[0])
hybrid_forecast = float(arima_forecast) + residual_correction
last_price = float(df['Close'].iloc[-1])
expected_change = (hybrid_forecast - last_price) / last_price
# Calculate volatility
returns = df['Close'].pct_change().dropna()
volatility = float(returns.rolling(window=20).std().iloc[-1])
# Generate signal
if expected_change > self.buy_threshold:
signal = "BUY"
emoji = "🟢"
elif expected_change < self.sell_threshold:
signal = "SELL"
emoji = "🔴"
else:
signal = "HOLD"
emoji = "🟡"
# Calculate confidence
confidence = min(0.9, max(0.1, abs(expected_change) / (volatility + 0.01)))
result = {
'ticker': ticker,
'last_price': last_price,
'arima_forecast': float(arima_forecast),
'residual_correction': residual_correction,
'hybrid_forecast': hybrid_forecast,
'expected_change': expected_change,
'volatility': volatility,
'confidence': confidence,
'signal': signal,
'emoji': emoji,
'timestamp': datetime.now(),
'analysis_time': datetime.now().isoformat()
}
# Cache the result
self.cache.save_result(ticker, result)
self.analyzed_tickers.add(ticker)
print(f"✅ {emoji} {ticker}: {signal} | Change: {expected_change:+.2%} | Confidence: {confidence:.1%}")
return result
except Exception as e:
print(f"❌ {ticker} analysis failed: {e}")
return None
def filter_results_by_signal(self, results: List[Dict[str, Any]],
signal_filter: Optional[str] = None,
min_confidence: Optional[float] = None) -> List[Dict[str, Any]]:
"""Filter results by signal type and confidence."""
if not results:
return results
filtered = results.copy()
# Filter by signal
if signal_filter:
signal_filter = signal_filter.upper()
filtered = [r for r in filtered if r['signal'] == signal_filter]
print(f"🎯 Filtered to {signal_filter} signals: {len(filtered)} stocks")
# Filter by confidence
if min_confidence:
filtered = [r for r in filtered if r['confidence'] >= min_confidence]
print(f"🎯 Filtered to {min_confidence:.1%}+ confidence: {len(filtered)} stocks")
return filtered
def analyze_all_stocks(self, signal_filter: Optional[str] = None,
min_confidence: Optional[float] = None,
save_results: bool = True) -> List[Dict[str, Any]]:
"""Analyze all unique stocks from all portfolios with optional filtering."""
all_stocks = self.get_all_unique_stocks()
print("🌐 ANALYZING ALL STOCKS")
print("=" * 50)
print(f"📊 Total unique stocks across all portfolios: {len(all_stocks)}")
if signal_filter:
print(f"🎯 Will filter for {signal_filter.upper()} signals")
if min_confidence:
print(f"🎯 Will filter for {min_confidence:.1%}+ confidence")
print(f"💾 Using smart caching to optimize analysis")
print()
# Analyze all stocks
results = []
failed_count = 0
for i, ticker in enumerate(all_stocks, 1):
print(f"[{i:3d}/{len(all_stocks)}] ", end="")
result = self.analyze_stock_cached(ticker)
if result:
results.append(result)
else:
failed_count += 1
print(f"\n📊 Analysis Complete:")
print(f" ✅ Successful: {len(results)} stocks")
print(f" ❌ Failed: {failed_count} stocks")
# Apply filters
if signal_filter or min_confidence:
print(f"\n🎯 Applying Filters:")
original_count = len(results)
results = self.filter_results_by_signal(results, signal_filter, min_confidence)
print(f" 📊 Results: {len(results)} of {original_count} stocks match criteria")
if not results:
print(f"❌ No stocks match the specified criteria")
return []
# Sort by confidence (highest first)
results.sort(key=lambda x: x['confidence'], reverse=True)
# Print filtered results
print(f"\n🏆 FILTERED RESULTS ({len(results)} stocks):")
print("=" * 60)
print(f"{'Rank':<4} {'Ticker':<8} {'Signal':<6} {'Change':<10} {'Confidence':<12} {'Price':<10}")
print("-" * 60)
for i, result in enumerate(results, 1):
print(f"{i:<4} {result['ticker']:<8} "
f"{result['emoji']} {result['signal']:<4} "
f"{result['expected_change']:<10.2%} "
f"{result['confidence']:<12.1%} "
f"${result['last_price']:<9.2f}")
# Summary statistics
if len(results) > 1:
avg_change = np.mean([r['expected_change'] for r in results])
avg_confidence = np.mean([r['confidence'] for r in results])
max_change = max([r['expected_change'] for r in results])
min_change = min([r['expected_change'] for r in results])
print(f"\n📊 Summary Statistics:")
print(f" Average Expected Change: {avg_change:+.2%}")
print(f" Average Confidence: {avg_confidence:.1%}")
print(f" Range: {min_change:+.2%} to {max_change:+.2%}")
# Save results
if save_results:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filter_suffix = ""
if signal_filter:
filter_suffix += f"_{signal_filter.lower()}"
if min_confidence:
filter_suffix += f"_conf{int(min_confidence*100)}"
filename = f"results/trading_signals/all_stocks{filter_suffix}_{timestamp}.csv"
# Create DataFrame and save
df_results = pd.DataFrame(results)
Path(filename).parent.mkdir(parents=True, exist_ok=True)
df_results.to_csv(filename, index=False)
print(f"\n📁 Results saved to: {filename}")
return results
def analyze_portfolio(self, portfolio_name: str) -> List[Dict[str, Any]]:
"""Analyze a single portfolio."""
if portfolio_name == "all":
return self.analyze_all_stocks()
tickers = self.config["tickers"].get(portfolio_name, [])
if not tickers:
print(f"❌ Portfolio '{portfolio_name}' not found in config")
return []
print(f"\n📊 Analyzing {portfolio_name} portfolio ({len(tickers)} stocks)...")
results = []
for ticker in tickers:
result = self.analyze_stock_cached(ticker)
if result:
results.append(result)
return results
def compare_portfolios(self, portfolio_names: List[str], save_results: bool = True) -> Dict[str, Any]:
"""Compare multiple portfolios with comprehensive analysis."""
print("🔬 SMART PORTFOLIO COMPARISON ANALYSIS")
print("=" * 60)
# Collect all unique tickers first
all_tickers = set()
portfolio_tickers = {}
for portfolio_name in portfolio_names:
tickers = self.config["tickers"].get(portfolio_name, [])
if tickers:
portfolio_tickers[portfolio_name] = tickers
all_tickers.update(tickers)
else:
print(f"⚠️ Portfolio '{portfolio_name}' not found in config")
print(f"📊 Analyzing {len(all_tickers)} unique stocks across {len(portfolio_tickers)} portfolios")
# Analyze each unique stock once
print(f"\n🔬 Stock Analysis Phase:")
stock_results = {}
for ticker in sorted(all_tickers):
result = self.analyze_stock_cached(ticker)
if result:
stock_results[ticker] = result
# Build portfolio results
print(f"\n📋 Portfolio Compilation Phase:")
portfolio_results = {}
portfolio_metrics = {}
for portfolio_name, tickers in portfolio_tickers.items():
print(f"\n📊 {portfolio_name.upper()} PORTFOLIO:")
print("-" * 50)
results = []
for ticker in tickers:
if ticker in stock_results:
result = stock_results[ticker]
results.append(result)
print(f" {result['emoji']} {ticker:<6}: {result['signal']:<4} ({result['expected_change']:+6.2%}, {result['confidence']:5.1%} conf)")
else:
print(f" ❌ {ticker}: Analysis failed")
portfolio_results[portfolio_name] = results
# Calculate portfolio metrics
metrics = self.comparator.calculate_portfolio_metrics(results, portfolio_name)
portfolio_metrics[portfolio_name] = metrics
# Print portfolio summary
if 'error' not in metrics:
print(f"\n 📈 Portfolio Summary:")
print(f" Expected Return: {metrics['expected_return']:+.2%}")
print(f" Quality Score: {metrics['quality_score']:.1f}/100")
print(f" Risk Level: {metrics['portfolio_volatility']:.2%}")
print(f" Action Signals: {metrics['buy_signals']} BUY, {metrics['sell_signals']} SELL")
# Perform comparison analysis
print(f"\n🏆 PORTFOLIO COMPARISON ANALYSIS:")
print("=" * 60)
comparison_results = self.comparator.compare_portfolios(portfolio_metrics)
if 'error' not in comparison_results:
# Print comparison table
print(f"\n📊 Portfolio Comparison Table:")
print(comparison_results['comparison_table'].to_string(index=False))
# Print rankings
print(f"\n🏆 Portfolio Rankings:")
rankings = comparison_results['rankings']
for category, top_portfolios in rankings.items():
print(f"\n {category.replace('_', ' ').title()}:")
for i, (portfolio, value) in enumerate(top_portfolios, 1):
if category in ['highest_return', 'expected_return']:
print(f" {i}. {portfolio}: {value:+.2%}")
elif 'volatility' in category or 'risk' in category:
print(f" {i}. {portfolio}: {value:.2%}")
else:
print(f" {i}. {portfolio}: {value:.2f}")
# Print recommendations
print(f"\n💡 Investment Recommendations:")
recommendations = comparison_results['recommendations']
for investor_type, recommendation in recommendations.items():
print(f" {investor_type.title()}: {recommendation}")
return {
'portfolio_results': portfolio_results,
'portfolio_metrics': portfolio_metrics,
'comparison_results': comparison_results,
'stock_results': stock_results
}
def parse_arguments():
"""Parse command line arguments with support for signal filtering."""
parser = argparse.ArgumentParser(
description='Smart Trading System with Portfolio Comparison and Signal Filtering',
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
python main.py # Show help and available portfolios
python main.py tech_giants # Analyze tech_giants portfolio
python main.py all # Analyze ALL stocks
python main.py all --buy # Show only BUY signals from all stocks
python main.py all --sell # Show only SELL signals from all stocks
python main.py all --buy --confidence 0.7 # BUY signals with 70%+ confidence
python main.py large_cap mid_cap # Compare two portfolios
python main.py compare conservative aggressive # Explicit portfolio comparison
python main.py tech_giants --buy # BUY signals from tech portfolio
"""
)
parser.add_argument('portfolios', nargs='*',
help='Portfolio names to analyze or compare (use "all" for all stocks)')
parser.add_argument('--buy', action='store_true',
help='Show only BUY signals')
parser.add_argument('--sell', action='store_true',
help='Show only SELL signals')
parser.add_argument('--hold', action='store_true',
help='Show only HOLD signals')
parser.add_argument('--confidence', type=float, metavar='THRESHOLD',
help='Minimum confidence threshold (0.0-1.0, e.g., 0.7 for 70%%)')
parser.add_argument('--compare', action='store_true',
help='Force comparison mode (auto-detected if multiple portfolios)')
parser.add_argument('--save', action='store_true', default=True,
help='Save results to CSV (default: True)')
parser.add_argument('--no-save', dest='save', action='store_false',
help='Do not save results to CSV')
return parser.parse_args()
def main():
"""Enhanced main function with argument parsing and signal filtering."""
args = parse_arguments()
system = SmartTradingSystem()
# Determine signal filter
signal_filter = None
if args.buy:
signal_filter = "BUY"
elif args.sell:
signal_filter = "SELL"
elif args.hold:
signal_filter = "HOLD"
# Handle different modes
if not args.portfolios:
# Show help and available portfolios
print("🚀 SMART TRADING SYSTEM WITH SIGNAL FILTERING")
print("=" * 60)
available_portfolios = list(system.config.get("tickers", {}).keys())
print(f"📊 Available portfolios ({len(available_portfolios)}):")
for i, portfolio in enumerate(available_portfolios, 1):
tickers = system.config["tickers"][portfolio]
print(f" {i:2d}. {portfolio:<20} ({len(tickers)} stocks)")
all_stocks = system.get_all_unique_stocks()
print(f"\n🌐 Total unique stocks across all portfolios: {len(all_stocks)}")
print("\n📖 Usage Examples:")
print(" python main.py all # Analyze all stocks")
print(" python main.py all --buy # Show only BUY signals")
print(" python main.py all --sell --confidence 0.8 # SELL signals with 80%+ confidence")
print(" python main.py tech_giants --buy # BUY signals from tech portfolio")
print(" python main.py large_cap mid_cap # Compare two portfolios")
return
# Handle "all" portfolio
if len(args.portfolios) == 1 and args.portfolios[0] == "all":
results = system.analyze_all_stocks(
signal_filter=signal_filter,
min_confidence=args.confidence,
save_results=args.save
)
return
# Handle portfolio comparison vs single portfolio
if len(args.portfolios) > 1 or args.compare:
# Multiple portfolios - comparison mode
portfolio_names = args.portfolios
print(f"🔬 Comparing portfolios: {', '.join(portfolio_names)}")
if signal_filter or args.confidence:
print("⚠️ Note: Signal filtering applies after portfolio comparison")
results = system.compare_portfolios(portfolio_names)
# Apply filtering to comparison results if requested
if signal_filter or args.confidence:
print(f"\n🎯 Applying filters to comparison results...")
for portfolio_name, portfolio_results in results['portfolio_results'].items():
filtered = system.filter_results_by_signal(
portfolio_results, signal_filter, args.confidence
)
if filtered:
print(f"\n{portfolio_name.upper()} - Filtered Results:")
for result in filtered:
print(f" {result['emoji']} {result['ticker']}: {result['signal']} "
f"({result['expected_change']:+.2%}, {result['confidence']:.1%})")
else:
print(f"\n{portfolio_name.upper()}: No stocks match filter criteria")
else:
# Single portfolio mode
portfolio_name = args.portfolios[0]
print(f"📊 Analyzing {portfolio_name} portfolio...")
results = system.analyze_portfolio(portfolio_name)
if results:
# Apply filtering
if signal_filter or args.confidence:
results = system.filter_results_by_signal(results, signal_filter, args.confidence)
if results:
print(f"\n📊 {portfolio_name.upper()} PORTFOLIO RESULTS:")
print("=" * 50)
# Sort by confidence
results.sort(key=lambda x: x['confidence'], reverse=True)
for result in results:
print(f"{result['emoji']} {result['ticker']:<6}: {result['signal']:<4} "
f"({result['expected_change']:+6.2%}, {result['confidence']:5.1%} conf)")
# Portfolio metrics
metrics = system.comparator.calculate_portfolio_metrics(results, portfolio_name)
if 'error' not in metrics:
print(f"\n📈 Portfolio Metrics:")
print(f" Expected Return: {metrics['expected_return']:+.2%}")
print(f" Quality Score: {metrics['quality_score']:.1f}/100")
print(f" Risk Level: {metrics['portfolio_volatility']:.2%}")
print(f" Buy Signals: {metrics['buy_signals']}")
print(f" Sell Signals: {metrics['sell_signals']}")
# Save filtered results if requested
if args.save and (signal_filter or args.confidence):
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filter_suffix = ""
if signal_filter:
filter_suffix += f"_{signal_filter.lower()}"
if args.confidence:
filter_suffix += f"_conf{int(args.confidence*100)}"
filename = f"results/trading_signals/{portfolio_name}{filter_suffix}_{timestamp}.csv"
df_results = pd.DataFrame(results)
Path(filename).parent.mkdir(parents=True, exist_ok=True)
df_results.to_csv(filename, index=False)
print(f"\n📁 Filtered results saved to: {filename}")
else:
print(f"❌ No stocks in {portfolio_name} match the filter criteria")
else:
print(f"❌ No valid results for {portfolio_name} portfolio")
if __name__ == "__main__":
main()