股票指标参数优化与回测系统 (cn-vhql/aiparameter)

股票指标参数优化与回测系统

https://github.com/cn-vhql/aiparameter
Admin
一个基于Python的股票技术指标参数优化与历史回测系统，通过算法搜索最优参数组合，辅助确定股票交易策略的最佳参数配置，并集成了完整的风险管理功能，支持13种技术指标的智能参数优化。

Tokens:11,618
Snippets:16
Trust Score:6.7
Update:6 months ago
Show doc for...
Context Summary (auto-generated)
Raw
# AI Parameter - Stock Technical Indicator Parameter Optimization and Backtesting System

AI Parameter is a Python-based system for optimizing technical indicator parameters and backtesting stock trading strategies. The system uses algorithmic search methods including grid search and genetic algorithms to find optimal parameter combinations for 13 different technical indicators. It integrates complete risk management functionality, supporting intelligent position sizing, automatic stop-loss/take-profit, and real-time risk assessment.

The system is built on a modern stack including Streamlit for the web interface, TA-Lib for technical indicator calculations, akshare for A-share market data fetching, and joblib for parallel computing. It provides a complete workflow from data acquisition, indicator calculation, parameter optimization, to backtesting with performance metrics calculation. The architecture separates concerns into data fetching, indicator calculation, optimization engines, backtesting engines, and risk management modules, making it highly extensible and maintainable.

## APIs and Key Functions

### Fetch Stock Data with Caching

```python
from data.data_fetcher import StockDataFetcher
from datetime import datetime

# Initialize fetcher with caching enabled (24-hour expiry)
fetcher = StockDataFetcher(cache_enabled=True, cache_dir="data_cache", cache_expiry_hours=24)

# Fetch stock data with automatic caching
df = fetcher.get_stock_data(
    stock_code="000001.SZ",      # Ping An Bank (Shenzhen)
    start_date="20230101",        # YYYYMMDD format
    end_date="20231231",
    period="日线",                # Daily data (日线/周线/月线)
    retry_count=3,                # Retry 3 times on failure
    use_cache=True                # Use cached data if available
)

if df is not None:
    print(f"Data shape: {df.shape}")
    print(f"Columns: {df.columns.tolist()}")
    # Output: ['open', 'high', 'low', 'close', 'volume', 'turnover', 'pct_change', 'amplitude']
    print(df.head())
else:
    print("Failed to fetch data")

# Get cache information
cache_info = fetcher.get_cache_info()
print(f"Cache files: {cache_info['cache_files']}")
print(f"Cache size: {cache_info['cache_size_mb']} MB")

# Clear specific cache
cleared_count = fetcher.clear_cache(stock_code="000001.SZ")
print(f"Cleared {cleared_count} cache files")
```

### Calculate Technical Indicators with Trading Signals

```python
from indicators.technical_indicators import TechnicalIndicators
import pandas as pd
import numpy as np

# Initialize indicator calculator
indicator_calc = TechnicalIndicators()

# Create sample data
dates = pd.date_range('2023-01-01', '2023-12-31', freq='D')
df = pd.DataFrame({
    'open': np.random.uniform(95, 105, len(dates)),
    'high': np.random.uniform(100, 110, len(dates)),
    'low': np.random.uniform(90, 100, len(dates)),
    'close': np.random.uniform(95, 105, len(dates)),
    'volume': np.random.randint(1000000, 10000000, len(dates))
}, index=dates)

# Calculate MACD with custom parameters
macd_params = {
    "fast_period": 12,
    "slow_period": 26,
    "signal_period": 9
}
df_macd = indicator_calc.calculate_indicator(df, "MACD", macd_params)
print(f"MACD columns: {[col for col in df_macd.columns if 'macd' in col.lower()]}")
print(f"Buy signals: {(df_macd['signal'] == 1).sum()}")
print(f"Sell signals: {(df_macd['signal'] == -1).sum()}")

# Calculate RSI with default parameters
df_rsi = indicator_calc.calculate_indicator(df, "RSI", {"timeperiod": 14})
print(f"RSI overbought signals (>70): {(df_rsi['rsi_14'] > 70).sum()}")
print(f"RSI oversold signals (<30): {(df_rsi['rsi_14'] < 30).sum()}")

# Calculate KDJ indicator
kdj_params = {"fastk_period": 9, "slowk_period": 3, "slowd_period": 3}
df_kdj = indicator_calc.calculate_indicator(df, "KDJ", kdj_params)
print(f"KDJ golden cross signals: {(df_kdj['signal'] == 1).sum()}")

# Get all available indicators
available_indicators = indicator_calc.get_available_indicators()
print(f"Supported indicators: {available_indicators}")
# Output: ['MACD', 'RSI', 'KDJ', '布林带', '均线', '乖离率', 'ATR', 'OBV', '威廉指标', 'CCI', 'ADX', '动量指标', '抛物线SAR']
```

### Run Backtesting with Risk Management

```python
from backtest.backtest_engine import BacktestEngine
from indicators.technical_indicators import TechnicalIndicators
import pandas as pd
import numpy as np

# Create sample data with signals
dates = pd.date_range('2023-01-01', '2023-12-31', freq='D')
df = pd.DataFrame({
    'close': np.cumsum(np.random.normal(0.001, 0.02, len(dates))) + 100,
    'open': np.cumsum(np.random.normal(0.001, 0.02, len(dates))) + 99,
    'high': np.cumsum(np.random.normal(0.001, 0.02, len(dates))) + 102,
    'low': np.cumsum(np.random.normal(0.001, 0.02, len(dates))) + 98,
    'volume': np.random.randint(1000000, 10000000, len(dates))
}, index=dates)

# Calculate indicator and generate signals
indicator_calc = TechnicalIndicators()
df_with_signals = indicator_calc.calculate_indicator(df, "MACD", {
    "fast_period": 12, "slow_period": 26, "signal_period": 9
})

# Initialize backtest engine with risk management enabled
engine = BacktestEngine(
    initial_capital=100000.0,
    commission_rate=0.0003,        # 0.03% commission
    slippage_rate=0.0001,          # 0.01% slippage
    enable_risk_management=True     # Enable risk control features
)

# Run backtest and get full results
results = engine.run_backtest(
    df_with_signals,
    signal_column='signal',
    price_column='close',
    return_full=True
)

# Display performance metrics
metrics = results['performance_metrics']
print(f"Annual Return: {metrics['年化收益率(%)']}%")
print(f"Max Drawdown: {metrics['最大回撤(%)']}%")
print(f"Sharpe Ratio: {metrics['夏普比率']}")
print(f"Win Rate: {metrics['胜率(%)']}%")
print(f"Total Trades: {metrics['总交易次数']}")
print(f"Net Profit: ${metrics['净利润']:,.2f}")

# Access equity curve
equity_curve = pd.DataFrame(results['equity_curve'])
print(f"Equity curve shape: {equity_curve.shape}")
print(equity_curve.head())

# Access trade records with risk information
trades = results['trades']
for trade in trades[:3]:  # Show first 3 trades
    print(f"{trade['date']}: {trade['action']} {trade['quantity']:.2f} @ ${trade['price']:.2f}")
    if 'risk_level' in trade:
        print(f"  Risk Level: {trade['risk_level']}")
    if 'sell_reason' in trade:
        print(f"  Reason: {trade['sell_reason']}")

# Generate text report
report = engine.generate_report(results)
print(report)
```

### Optimize Parameters with Grid Search

```python
from optimization.optimizer import ParameterOptimizer
from data.data_fetcher import StockDataFetcher
import pandas as pd

# Fetch real stock data
fetcher = StockDataFetcher()
df = fetcher.get_stock_data("000001.SZ", "20230101", "20231231", "日线")

# Initialize optimizer with parallel processing
optimizer = ParameterOptimizer(n_jobs=-1)  # Use all CPU cores

# Define parameter search ranges
param_ranges = {
    "fast_period": (5, 20),      # Fast period from 5 to 20
    "slow_period": (15, 40),     # Slow period from 15 to 40
    "signal_period": (3, 15)     # Signal period from 3 to 15
}

# Progress callback function
def progress_callback(current, total):
    progress = int(current / total * 100)
    print(f"Progress: {progress}% ({current}/{total} combinations)")

# Run grid search optimization
best_params, results_df = optimizer.optimize_parameters(
    df=df,
    indicator_name="MACD",
    algorithm="grid_search",              # or "网格搜索"
    param_ranges=param_ranges,
    objective="年化收益率",               # Optimize for annual return
    progress_callback=progress_callback
)

# Display optimization results
print(f"Best parameters: {best_params}")
print(f"Best annual return: {results_df['年化收益率'].max():.2f}%")

# Show top 5 parameter combinations
top_5 = results_df.nlargest(5, '年化收益率')
print("\nTop 5 combinations:")
for idx, row in top_5.iterrows():
    print(f"Params: {row[list(param_ranges.keys())].to_dict()}")
    print(f"  Annual Return: {row['年化收益率']:.2f}%")
    print(f"  Max Drawdown: {row['最大回撤']:.2f}%")
    print(f"  Sharpe Ratio: {row['夏普比率']:.2f}")
    print(f"  Trades: {row['交易次数']}\n")

# Get best equity curve for visualization
best_equity_curve = optimizer.get_best_equity_curve()
if best_equity_curve is not None:
    print(f"Best equity curve shape: {best_equity_curve.shape}")
```

### Optimize Parameters with Genetic Algorithm

```python
from optimization.optimizer import ParameterOptimizer
from data.data_fetcher import StockDataFetcher

# Fetch stock data
fetcher = StockDataFetcher()
df = fetcher.get_stock_data("600036.SH", "20220101", "20231231", "日线")

# Initialize optimizer
optimizer = ParameterOptimizer(n_jobs=-1)

# Define parameter ranges (min, max tuples)
param_ranges = {
    "timeperiod": (6, 30)  # RSI period from 6 to 30
}

# Progress callback
def show_progress(generation, total_generations):
    print(f"Generation {generation}/{total_generations}")

# Run genetic algorithm optimization
best_params, results_df = optimizer.optimize_parameters(
    df=df,
    indicator_name="RSI",
    algorithm="genetic",                  # or "遗传算法"
    param_ranges=param_ranges,
    population_size=30,                   # Population size
    generations=10,                       # Number of generations
    objective="夏普比率",                 # Optimize for Sharpe ratio
    progress_callback=show_progress
)

print(f"Genetic algorithm found best parameters: {best_params}")
print(f"Best Sharpe ratio: {results_df['夏普比率'].max():.2f}")

# Analyze genetic algorithm convergence
if not results_df.empty and 'generation' in results_df.columns:
    convergence = results_df.groupby('generation')['夏普比率'].agg(['max', 'mean', 'std'])
    print("\nConvergence analysis:")
    print(convergence)
```

### Risk Management Configuration

```python
from risk_management.risk_manager import RiskManager
from backtest.backtest_engine import BacktestEngine

# Initialize risk manager with custom parameters
risk_manager = RiskManager(
    max_position_size=1.0,         # Max 100% position
    max_loss_per_trade=0.02,       # Max 2% loss per trade
    max_drawdown=0.15,             # Max 15% drawdown limit
    stop_loss_pct=0.05,            # 5% stop loss
    take_profit_pct=0.10,          # 10% take profit
    max_daily_loss=0.03            # Max 3% daily loss
)

# Calculate position size based on risk
position_size = risk_manager.calculate_position_size(
    available_capital=100000,
    entry_price=100.0,
    stop_loss_price=95.0,          # 5% stop loss
    risk_per_share=None            # Auto-calculate from prices
)
print(f"Recommended position size: {position_size:.2f} shares")
print(f"Investment amount: ${position_size * 100:.2f}")

# Check if entry is allowed
can_enter, reason = risk_manager.check_entry_risk(
    current_price=100.0,
    entry_price=100.0,
    stop_loss_price=95.0,
    current_equity=100000
)
print(f"Entry allowed: {can_enter}, Reason: {reason}")

# Calculate stop loss and take profit prices
stop_loss = risk_manager.calculate_stop_loss(entry_price=100.0, method="fixed")
take_profit = risk_manager.calculate_take_profit(entry_price=100.0, method="risk_reward")
print(f"Stop loss: ${stop_loss:.2f}, Take profit: ${take_profit:.2f}")

# Calculate risk-reward ratio
risk_reward = risk_manager.calculate_risk_reward_ratio(
    entry_price=100.0,
    stop_loss_price=95.0
)
print(f"Risk-reward ratio: 1:{risk_reward:.2f}")

# Assess risk level for a trade
risk_level = risk_manager.assess_risk_level(
    available_cash=100000,
    buy_quantity=500,
    current_price=100.0
)
print(f"Risk level: {risk_level}")

# Update risk metrics during trading
risk_manager.update_risk_metrics(current_equity=95000, daily_pnl=-2000)

# Get comprehensive risk report
risk_report = risk_manager.get_risk_report()
print("\nRisk Report:")
for key, value in risk_report.items():
    print(f"  {key}: {value}")

# Check for violations
has_violation = risk_manager.check_violation(
    current_equity=95000,
    initial_capital=100000
)
if has_violation:
    print("Risk violation detected!")

# Adjust risk parameters based on market volatility
risk_manager.adjust_risk_parameters(market_volatility=0.035)
```

### Complete End-to-End Trading Strategy Optimization

```python
from data.data_fetcher import StockDataFetcher
from indicators.technical_indicators import TechnicalIndicators
from optimization.optimizer import ParameterOptimizer
from backtest.backtest_engine import BacktestEngine
import pandas as pd

# Step 1: Fetch stock data with caching
fetcher = StockDataFetcher(cache_enabled=True, cache_expiry_hours=24)
df = fetcher.get_stock_data("000001.SZ", "20220101", "20231231", "日线")

if df is None or df.empty:
    raise ValueError("Failed to fetch data")

print(f"Loaded {len(df)} trading days")

# Step 2: Define optimization parameters
param_ranges = {
    "timeperiod": (10, 30),        # Bollinger Bands period
    "nbdevup": (1, 3),             # Upper band std deviation
    "nbdevdn": (1, 3)              # Lower band std deviation
}

# Step 3: Run optimization with progress tracking
optimizer = ParameterOptimizer(n_jobs=-1)

progress_history = []
def track_progress(current, total):
    percent = int(current / total * 100)
    progress_history.append(percent)
    if current % 100 == 0 or current == total:
        print(f"Optimization: {percent}% complete")

best_params, results_df = optimizer.optimize_parameters(
    df=df,
    indicator_name="布林带",
    algorithm="grid_search",
    param_ranges=param_ranges,
    objective="夏普比率",
    progress_callback=track_progress
)

# Step 4: Analyze results
print(f"\nOptimization complete!")
print(f"Best parameters: {best_params}")
print(f"Tested {len(results_df)} combinations")

# Statistical analysis of results
print("\nPerformance distribution:")
print(results_df['夏普比率'].describe())

# Step 5: Validate best parameters with risk management
indicator_calc = TechnicalIndicators()
df_validated = indicator_calc.calculate_indicator(df.copy(), "布林带", best_params)

engine = BacktestEngine(
    initial_capital=100000,
    enable_risk_management=True
)

validation_results = engine.run_backtest(df_validated, return_full=True)

# Step 6: Generate comprehensive report
print("\n" + "="*60)
print("VALIDATION RESULTS WITH RISK MANAGEMENT")
print("="*60)
metrics = validation_results['performance_metrics']
for key, value in metrics.items():
    print(f"{key}: {value}")

# Step 7: Export results for further analysis
results_df.to_csv("optimization_results.csv", index=False)
print(f"\nResults exported to optimization_results.csv")

# Step 8: Get equity curve for visualization
equity_curve = optimizer.get_best_equity_curve()
if equity_curve is not None:
    equity_curve.to_csv("equity_curve.csv")
    print("Equity curve exported to equity_curve.csv")
```

### Streamlit Web Application Entry Point

```python
# Run the complete web application
# Terminal command:
# streamlit run app.py

# The app.py provides:
# - Interactive parameter configuration via sidebar
# - 13 technical indicators selection with descriptions
# - Real-time optimization progress display
# - Visual backtesting results with Plotly charts
# - Risk management toggle and detailed explanations
# - Grid search and genetic algorithm options
# - Automatic parameter validation
# - Performance metrics dashboard
# - Trade history and equity curve visualization

# Access the application at http://localhost:8501
# Configure stock code (e.g., 000001.SZ), time period, indicator, and parameters
# Click "🚀 开始参数优化" to start optimization
# View results including best parameters, equity curves, and detailed metrics
```

## Summary and Integration

AI Parameter provides a comprehensive solution for systematic trading strategy development on A-share markets. The system excels at automating the tedious process of parameter tuning for technical indicators through intelligent optimization algorithms. Users can leverage grid search for exhaustive parameter space exploration or genetic algorithms for efficient optimization of complex parameter combinations. The integrated risk management system ensures that optimized strategies maintain acceptable risk profiles through position sizing, stop-loss/take-profit mechanisms, and real-time risk assessment.

The modular architecture allows for easy extension with new indicators, optimization algorithms, or risk management strategies. The caching mechanism significantly improves performance for repeated data fetches, while parallel computing capabilities enable efficient processing of large parameter spaces. The Streamlit-based interface makes the system accessible to both traders and developers, providing immediate visual feedback on strategy performance. All components are designed to work seamlessly together, from data fetching through optimization to final backtesting, creating a complete pipeline for quantitative trading strategy development and validation.