引言：为什么需要懒人版资产配置策略？
在现代投资世界中，资产配置是决定长期收益的关键因素。研究表明，超过90%的投资回报差异来自于资产配置而非个股选择。然而，传统的资产配置需要投资者持续监控市场、手动调整持仓，这对大多数人来说既耗时又复杂。
懒人版资产配置策略的核心理念是”设置一次，长期自动运行”。通过预设规则和自动化工具，投资者可以实现：
- 省时省力：无需每日盯盘，每月或每季度只需几分钟检查
- 纪律性执行：避免情绪化交易，严格执行既定策略
- 风险分散：通过多元化配置降低单一资产风险
- 成本可控：利用低成本ETF和自动化交易降低交易费用
本文将从零基础开始，详细解析如何构建一个完整的自动调仓系统，包括策略设计、工具选择、代码实现和实战操作。
第一部分：基础概念理解
1.1 什么是资产配置？
资产配置是指将投资资金分配到不同类型的资产类别中，以平衡风险和收益。常见的资产类别包括：
股票类资产：
- 宽基指数ETF（如沪深300、标普500）
- 行业主题ETF（如科技、消费、医药）
- 个股（风险较高，不适合懒人策略）
债券类资产：
- 国债ETF
- 企业债ETF
- 可转债
现金类资产：
- 货币基金
- 短期国债
- 银行存款
另类资产：
- 黄金ETF
- 房地产REITs
- 大宗商品
1.2 懒人策略的核心原则
原则一：简单至上
- 选择3-5个资产类别即可，不要过度复杂化
- 优先选择宽基指数ETF，避免行业轮动陷阱
原则二：固定比例
- 预设固定的配置比例（如股6债4）
- 只有在极端市场情况下才临时调整
原则三：定期再平衡
- 设定固定时间间隔（如每季度）
- 或设定偏离阈值（如某类资产偏离目标比例5%）
原则四：低成本执行
- 选择管理费率低的ETF
- 利用自动化工具减少交易成本
第二部分：策略设计与选择
2.1 经典懒人策略介绍
策略A：永久组合（Permanent Portfolio）
由Harry Browne提出，包含四类资产各25%：
- 股票：追求增长
- 长期国债：应对通缩
- 黄金：应对通胀
- 现金：应对经济衰退
优点：在任何经济环境下都能保持稳定 缺点：牛市时收益相对较低
策略B：60/40经典组合
- 60%股票指数ETF
- 40%债券指数ETF
优点：简单有效，历史表现优秀 缺点：在通胀高企时期债券表现不佳
策略C：全天候组合（All Weather）
由Ray Dalio提出，根据风险平价原则配置：
- 30%股票
- 40%长期国债
- 15%中期国债
- 7.5%黄金
- 7.5%大宗商品
优点：风险分散极佳 缺点：需要更多资产类别，操作稍复杂
2.2 策略选择指南
新手推荐：从60/40组合开始，最容易理解和执行 进阶选择：永久组合，适合追求极致稳定的投资者 专业选择：全天候组合，需要更多监控和资金
2.3 自动调仓的触发条件
时间触发：
- 每月/每季度/每半年固定日期
- 优点：简单，无需监控
- 缺点：可能错过最佳调仓时机
阈值触发：
- 当某类资产偏离目标比例超过X%时
- 通常X=5%或10%
- 优点：更灵活，减少不必要的交易
- 缺点：需要持续监控
混合触发：
- 时间+阈值结合
- 例如：每季度检查，但只有偏离超过5%时才调仓
第三部分：工具与平台选择
3.1 券商平台选择
国内券商：
- 华泰证券、东方财富、招商证券等
- 支持条件单、网格交易等自动化功能
- 佣金费率：万1到万3
国际券商：
- Interactive Brokers（IBKR）
- 支持全球资产配置
- 提供API接口，适合程序化交易
3.2 自动化工具
方案一：券商条件单（适合非程序员）
华泰证券条件单示例：
设置步骤：
1. 登录华泰证券APP
2. 进入"交易" -> "条件单"
3. 选择"时间条件单"
4. 设置触发时间：每月第一个交易日 9:30
5. 设置买入/卖出标的和数量
6. 提交并确认
优点：无需编程，操作简单
缺点：灵活性有限，无法处理复杂逻辑
方案二：Python脚本（适合程序员）
核心优势：
- 完全自定义策略逻辑
- 可集成多个数据源
- 支持复杂计算和风控
- 可扩展性强
所需库：
# 核心库
import pandas as pd # 数据处理
import numpy as np # 数值计算
import yfinance as yf # 获取金融数据
import akshare as ak # 国内数据源
import time # 时间控制
import logging # 日志记录
# 券商API（以东方财富为例）
# 需要安装东方财富客户端并开通API权限
from东方财富API import EastMoneyAPI
方案三：第三方平台（折中方案）
聚宽（JoinQuant）：
- 提供在线策略编写和回测
- 支持实盘交易对接
- 月费约200-500元
米筐（RiceQuant）：
- 类似聚宽，功能更丰富
- 支持多种编程语言
- 月费约300-600元
第四部分：从零开始的实战代码实现
4.1 环境准备
安装必要库：
pip install pandas numpy yfinance akshare matplotlib
# 如果使用东方财富API，需要额外安装
pip install eastmoney-api
创建项目结构：
auto_rebalance/
├── config.py # 配置文件
├── data_fetcher.py # 数据获取模块
├── calculator.py # 计算模块
├── trader.py # 交易执行模块
├── main.py # 主程序
├── logger.py # 日志模块
└── requirements.txt # 依赖列表
4.2 配置文件（config.py）
# config.py
"""
懒人资产配置策略配置文件
"""
# 策略配置
STRATEGY_CONFIG = {
'strategy_name': '60/40经典组合',
'target_allocation': {
'股票': 0.60,
'债券': 0.40
},
'rebalance_trigger': {
'type': 'mixed', # 'time', 'threshold', 'mixed'
'time_interval': 'quarterly', # 'monthly', 'quarterly', 'semi_annual', 'annual'
'threshold': 0.05, # 5%偏离阈值
'check_date': 'first_trading_day' # 每月第一个交易日
}
}
# 资产配置详情
ASSET_CONFIG = {
'股票': {
'symbol': '510300', # 华泰柏瑞沪深300ETF
'name': '沪深300ETF',
'market': 'sh',
'target_ratio': 0.60
},
'债券': {
'symbol': '511010', # 国泰上证5年期国债ETF
'name': '国债ETF',
'market': 'sh',
'target_ratio': 0.40
}
}
# 交易参数
TRADING_PARAMS = {
'total_capital': 100000, # 总资金（元）
'min_trade_amount': 100, # 最小交易金额（元）
'commission_rate': 0.0001, # 佣金率（万1）
'slippage': 0.001, # 滑点（千1）
'max_single_trade': 50000, # 单笔最大交易额
'trading_hours': {
'start': '09:30',
'end': '15:00'
}
}
# 账户信息（实际使用时需要真实信息）
ACCOUNT_INFO = {
'broker': 'eastmoney', # 'eastmoney', 'huatai', 'interactive_brokers'
'account_id': 'your_account_id',
'api_key': 'your_api_key',
'api_secret': 'your_api_secret'
}
# 日志配置
LOG_CONFIG = {
'level': 'INFO',
'file': 'auto_rebalance.log',
'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
}
4.3 数据获取模块（data_fetcher.py）
# data_fetcher.py
"""
数据获取模块
支持实时价格和历史数据获取
"""
import yfinance as yf
import akshare as ak
import pandas as pd
import time
import logging
logger = logging.getLogger(__name__)
class DataFetcher:
def __init__(self):
self.retry_times = 3
self.timeout = 10
def get_realtime_price(self, symbol, market='sh'):
"""
获取实时价格（使用AkShare）
"""
for attempt in range(self.retry_times):
try:
if market == 'sh':
# 沪市股票
ticker = f"{symbol}.SH"
else:
# 深市股票
ticker = f"{symbol}.SZ"
# 使用AkShare获取实时行情
df = ak.stock_zh_a_spot_em()
stock_data = df[df['代码'] == symbol]
if not stock_data.empty:
price = stock_data['最新价'].values[0]
logger.info(f"获取{symbol}实时价格: {price}")
return float(price)
else:
logger.warning(f"未找到{symbol}的数据")
return None
except Exception as e:
logger.error(f"获取{symbol}价格失败: {e}")
time.sleep(1)
return None
def get_realtime_price_yfinance(self, symbol, market='sh'):
"""
使用yfinance获取价格（主要用于国际资产）
"""
try:
# 国内ETF需要特殊处理
if market == 'sh':
ticker = f"{symbol}.SS"
elif market == 'sz':
ticker = f"{symbol}.SZ"
else:
ticker = symbol
stock = yf.Ticker(ticker)
# 获取最近一天的收盘价
hist = stock.history(period="1d")
if not hist.empty:
price = hist['Close'].iloc[-1]
logger.info(f"YFinance获取{symbol}价格: {price}")
return float(price)
else:
logger.warning(f"YFinance未找到{symbol}的数据")
return None
except Exception as e:
logger.error(f"YFinance获取{symbol}价格失败: {e}")
return None
def get_historical_data(self, symbol, period="1y", market='sh'):
"""
获取历史数据用于分析
"""
try:
if market == 'sh':
ticker = f"{symbol}.SS"
else:
ticker = f"{symbol}.SZ"
stock = yf.Ticker(ticker)
hist = stock.history(period=period)
if not hist.empty:
logger.info(f"获取{symbol}历史数据成功，共{len(hist)}条")
return hist
else:
logger.warning(f"未找到{symbol}的历史数据")
return None
except Exception as e:
logger.error(f"获取{symbol}历史数据失败: {e}")
return None
def get_current_positions(self, account_api):
"""
获取当前持仓（需要券商API）
"""
try:
# 这里需要根据具体券商API实现
# 以伪代码形式展示
positions = account_api.get_positions()
# 格式化为字典
position_dict = {}
for pos in positions:
symbol = pos['symbol']
shares = pos['shares']
market_value = pos['market_value']
position_dict[symbol] = {
'shares': shares,
'market_value': market_value,
'current_price': pos['current_price']
}
logger.info(f"当前持仓: {position_dict}")
return position_dict
except Exception as e:
logger.error(f"获取持仓失败: {e}")
return {}
def check_trading_time(self):
"""
检查当前是否为交易时间
"""
from datetime import datetime
now = datetime.now()
# 检查是否为交易日
if now.weekday() >= 5: # 周六周日
return False
current_time = now.strftime("%H:%M")
trading_hours = TRADING_PARAMS['trading_hours']
if trading_hours['start'] <= current_time <= trading_hours['end']:
return True
else:
return False
# 测试代码
if __name__ == "__main__":
fetcher = DataFetcher()
# 测试获取价格
price = fetcher.get_realtime_price('510300', 'sh')
print(f"沪深300ETF价格: {price}")
4.4 计算模块（calculator.py）
# calculator.py
"""
计算模块
负责计算调仓指令
"""
import pandas as pd
import numpy as np
import logging
logger = logging.getLogger(__name__)
class RebalanceCalculator:
def __init__(self, config):
self.config = config
self.trading_params = config['TRADING_PARAMS']
self.asset_config = config['ASSET_CONFIG']
self.strategy_config = config['STRATEGY_CONFIG']
def calculate_target_shares(self, current_prices, total_capital):
"""
计算目标持仓数量
"""
target_shares = {}
for asset_name, asset_info in self.asset_config.items():
target_ratio = asset_info['target_ratio']
symbol = asset_info['symbol']
price = current_prices.get(symbol)
if price is None:
logger.error(f"无法获取{symbol}的价格，跳过该资产计算")
continue
# 计算目标金额
target_amount = total_capital * target_ratio
# 计算目标股数（取整）
target_share = int(target_amount / price)
target_shares[symbol] = {
'name': asset_info['name'],
'target_shares': target_share,
'target_amount': target_share * price,
'current_price': price,
'target_ratio': target_ratio
}
logger.info(f"{asset_info['name']}目标持仓: {target_share}股，金额: {target_share * price:.2f}元")
return target_shares
def calculate_rebalance_trades(self, current_positions, target_shares):
"""
计算调仓交易指令
"""
trades = []
for symbol, target_info in target_shares.items():
# 获取当前持仓
current_share = current_positions.get(symbol, {}).get('shares', 0)
current_price = target_info['current_price']
target_share = target_info['target_shares']
# 计算需要调整的数量
share_diff = target_share - current_share
if share_diff == 0:
logger.info(f"{target_info['name']}持仓已达标，无需调整")
continue
# 计算交易金额
trade_amount = abs(share_diff) * current_price
# 检查最小交易金额
if trade_amount < self.trading_params['min_trade_amount']:
logger.warning(f"交易金额{trade_amount:.2f}小于最小交易金额，跳过")
continue
# 检查最大单笔交易限制
if trade_amount > self.trading_params['max_single_trade']:
logger.warning(f"交易金额{trade_amount:.2f}超过单笔最大限制，需要分批交易")
# 分批处理逻辑（简化版）
batch_count = int(np.ceil(trade_amount / self.trading_params['max_single_trade']))
batch_share = int(share_diff / batch_count)
for i in range(batch_count):
if i == batch_count - 1:
# 最后一批处理剩余
batch_share = share_diff - batch_share * i
if batch_share != 0:
trades.append({
'symbol': symbol,
'name': target_info['name'],
'action': 'BUY' if batch_share > 0 else 'SELL',
'shares': abs(batch_share),
'price': current_price,
'amount': abs(batch_share) * current_price,
'reason': f'批量调整 {i+1}/{batch_count}'
})
continue
# 确定买卖方向
action = 'BUY' if share_diff > 0 else 'SELL'
trades.append({
'symbol': symbol,
'name': target_info['name'],
'action': action,
'shares': abs(share_diff),
'price': current_price,
'amount': trade_amount,
'reason': '定期再平衡'
})
logger.info(f"生成交易指令: {action} {target_info['name']} {abs(share_diff)}股，金额: {trade_amount:.2f}元")
return trades
def check_rebalance_needed(self, current_positions, current_prices, total_capital):
"""
检查是否需要调仓（阈值触发）
"""
# 计算当前持仓比例
total_value = sum(pos['market_value'] for pos in current_positions.values())
if total_value == 0:
return True, "初始建仓"
current_ratios = {}
for symbol, pos in current_positions.items():
current_ratios[symbol] = pos['market_value'] / total_value
# 计算目标比例
target_ratios = {info['symbol']: info['target_ratio'] for info in self.asset_config.values()}
# 检查偏离度
max_deviation = 0
for symbol in target_ratios:
current_ratio = current_ratios.get(symbol, 0)
target_ratio = target_ratios[symbol]
deviation = abs(current_ratio - target_ratio)
max_deviation = max(max_deviation, deviation)
threshold = self.strategy_config['rebalance_trigger']['threshold']
if max_deviation > threshold:
logger.info(f"最大偏离度: {max_deviation:.2%}，超过阈值{threshold:.2%}，需要调仓")
return True, f"偏离阈值触发: {max_deviation:.2%}"
else:
logger.info(f"最大偏离度: {max_deviation:.2%}，未超过阈值{threshold:.2%}，无需调仓")
return False, "未达到阈值"
def calculate_total_cost(self, trades):
"""
计算总交易成本
"""
total_commission = 0
total_slippage = 0
for trade in trades:
# 佣金
commission = trade['amount'] * self.trading_params['commission_rate']
# 最低佣金（通常5元）
commission = max(commission, 5)
total_commission += commission
# 滑点成本
slippage = trade['amount'] * self.trading_params['slippage']
total_slippage += slippage
total_cost = total_commission + total_slippage
return {
'total_cost': total_cost,
'commission': total_commission,
'slippage': total_slippage,
'cost_ratio': total_cost / sum(t['amount'] for t in trades) if trades else 0
}
# 测试代码
if __name__ == "__main__":
from config import STRATEGY_CONFIG, ASSET_CONFIG, TRADING_PARAMS
config = {
'STRATEGY_CONFIG': STRATEGY_CONFIG,
'ASSET_CONFIG': ASSET_CONFIG,
'TRADING_PARAMS': TRADING_PARAMS
}
calculator = RebalanceCalculator(config)
# 模拟数据
current_prices = {'510300': 3.5, '511010': 100}
total_capital = 100000
target_shares = calculator.calculate_target_shares(current_prices, total_capital)
print("目标持仓:", target_shares)
# 模拟当前持仓
current_positions = {
'510300': {'shares': 15000, 'market_value': 52500, 'current_price': 3.5},
'511010': {'shares': 400, 'market_value': 40000, 'current_price': 100}
}
trades = calculator.calculate_rebalance_trades(current_positions, target_shares)
print("交易指令:", trades)
4.5 交易执行模块（trader.py）
# trader.py
"""
交易执行模块
负责实际下单操作
"""
import logging
import time
from datetime import datetime
logger = logging.getLogger(__name__)
class Trader:
def __init__(self, config):
self.config = config
self.account_info = config['ACCOUNT_INFO']
self.trading_params = config['TRADING_PARAMS']
self.api = None
self.initialize_api()
def initialize_api(self):
"""
初始化券商API
"""
broker = self.account_info['broker']
if broker == 'eastmoney':
# 东方财富API（需要实际安装和配置）
try:
from eastmoney_api import EastMoneyAPI
self.api = EastMoneyAPI(
account_id=self.account_info['account_id'],
api_key=self.account_info['api_key'],
api_secret=self.account_info['api_secret']
)
logger.info("东方财富API初始化成功")
except ImportError:
logger.warning("未安装东方财富API，使用模拟模式")
self.api = None
elif broker == 'huatai':
# 华泰证券API（示例）
logger.warning("华泰证券API需要单独配置，使用模拟模式")
self.api = None
elif broker == 'interactive_brokers':
# IB API（示例）
logger.warning("IB API需要单独配置，使用模拟模式")
self.api = None
else:
logger.warning("未知券商，使用模拟模式")
self.api = None
def execute_trade(self, trade, simulate=True):
"""
执行单个交易
"""
symbol = trade['symbol']
action = trade['action']
shares = trade['shares']
price = trade['price']
amount = trade['amount']
logger.info(f"{'[模拟]' if simulate else '[实盘]'} 执行交易: {action} {symbol} {shares}股，价格: {price}，金额: {amount:.2f}元")
if simulate:
# 模拟模式，只记录不执行
logger.info(f"模拟交易成功: {action} {symbol}")
return {'status': 'simulated', 'order_id': f'SIM_{int(time.time())}'}
else:
# 实盘模式（需要真实API）
if self.api is None:
logger.error("API未初始化，无法执行实盘交易")
return {'status': 'failed', 'error': 'API not initialized'}
try:
# 根据买卖方向调用不同API
if action == 'BUY':
order_result = self.api.place_buy_order(
symbol=symbol,
shares=shares,
price=price
)
else:
order_result = self.api.place_sell_order(
symbol=symbol,
shares=shares,
price=price
)
logger.info(f"实盘交易成功: {order_result}")
return {'status': 'success', 'order_id': order_result.get('order_id')}
except Exception as e:
logger.error(f"实盘交易失败: {e}")
return {'status': 'failed', 'error': str(e)}
def execute_rebalance(self, trades, simulate=True, delay=1):
"""
执行调仓（批量交易）
"""
if not trades:
logger.info("没有需要执行的交易")
return []
results = []
# 先卖后买（避免资金占用）
sell_trades = [t for t in trades if t['action'] == 'SELL']
buy_trades = [t for t in trades if t['action'] == 'BUY']
sorted_trades = sell_trades + buy_trades
for i, trade in enumerate(sorted_trades):
logger.info(f"执行第 {i+1}/{len(sorted_trades)} 笔交易")
result = self.execute_trade(trade, simulate=simulate)
results.append(result)
# 交易间隔，避免过于频繁
if delay > 0 and i < len(sorted_trades) - 1:
time.sleep(delay)
return results
def check_order_status(self, order_id):
"""
检查订单状态（实盘模式）
"""
if self.api is None:
return {'status': 'simulated'}
try:
status = self.api.get_order_status(order_id)
return status
except Exception as e:
logger.error(f"查询订单状态失败: {e}")
return {'status': 'error', 'error': str(e)}
def get_account_balance(self):
"""
获取账户资金余额
"""
if self.api is None:
# 模拟返回
return {
'cash': 50000,
'market_value': 100000,
'total_assets': 150000
}
try:
balance = self.api.get_balance()
return balance
except Exception as e:
logger.error(f"获取账户余额失败: {e}")
return None
# 测试代码
if __name__ == "__main__":
from config import ACCOUNT_INFO, TRADING_PARAMS
config = {
'ACCOUNT_INFO': ACCOUNT_INFO,
'TRADING_PARAMS': TRADING_PARAMS
}
trader = Trader(config)
# 模拟交易
test_trade = {
'symbol': '510300',
'name': '沪深300ETF',
'action': 'BUY',
'shares': 1000,
'price': 3.5,
'amount': 3500,
'reason': '测试交易'
}
result = trader.execute_trade(test_trade, simulate=True)
print("交易结果:", result)
4.6 日志模块（logger.py）
# logger.py
"""
日志模块
"""
import logging
import sys
from datetime import datetime
def setup_logging(log_file='auto_rebalance.log', level='INFO'):
"""
配置日志系统
"""
# 创建日志器
logger = logging.getLogger('AutoRebalance')
logger.setLevel(getattr(logging, level.upper()))
# 清除已有handler
logger.handlers.clear()
# 文件处理器
file_handler = logging.FileHandler(log_file, encoding='utf-8')
file_formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
file_handler.setFormatter(file_formatter)
logger.addHandler(file_handler)
# 控制台处理器
console_handler = logging.StreamHandler(sys.stdout)
console_formatter = logging.Formatter(
'%(asctime)s - %(levelname)s - %(message)s'
)
console_handler.setFormatter(console_formatter)
logger.addHandler(console_handler)
return logger
# 全局日志实例
logger = setup_logging()
4.7 主程序（main.py）
# main.py
"""
主程序
整合所有模块，实现自动调仓
"""
import sys
import time
from datetime import datetime
import logging
# 导入自定义模块
from config import *
from data_fetcher import DataFetcher
from calculator import RebalanceCalculator
from trader import Trader
from logger import setup_logging
class AutoRebalanceSystem:
def __init__(self, config, simulate=True):
"""
初始化自动调仓系统
"""
self.config = config
self.simulate = simulate
self.logger = setup_logging(
log_file=config['LOG_CONFIG']['file'],
level=config['LOG_CONFIG']['level']
)
# 初始化各模块
self.fetcher = DataFetcher()
self.calculator = RebalanceCalculator(config)
self.trader = Trader(config)
self.logger.info("="*50)
self.logger.info("懒人资产配置自动调仓系统启动")
self.logger.info(f"策略: {config['STRATEGY_CONFIG']['strategy_name']}")
self.logger.info(f"模式: {'模拟' if simulate else '实盘'}")
self.logger.info("="*50)
def check_trigger_condition(self):
"""
检查触发条件
"""
trigger_type = self.config['STRATEGY_CONFIG']['rebalance_trigger']['type']
if trigger_type == 'time':
# 时间触发
return self._check_time_trigger()
elif trigger_type == 'threshold':
# 阈值触发
return self._check_threshold_trigger()
elif trigger_type == 'mixed':
# 混合触发
time_ok = self._check_time_trigger()
threshold_ok = self._check_threshold_trigger()
return time_ok or threshold_ok
else:
self.logger.error(f"未知触发类型: {trigger_type}")
return False
def _check_time_trigger(self):
"""
检查时间触发条件
"""
now = datetime.now()
time_interval = self.config['STRATEGY_CONFIG']['rebalance_trigger']['time_interval']
check_date = self.config['STRATEGY_CONFIG']['rebalance_trigger']['check_date']
# 检查是否为交易日
if now.weekday() >= 5:
return False
# 检查时间间隔
if time_interval == 'monthly':
# 每月第一个交易日
if check_date == 'first_trading_day':
# 判断是否为当月第一个交易日
first_day = now.replace(day=1)
if first_day.weekday() >= 5:
# 如果是周末，取下一个周一
days_to_monday = (7 - first_day.weekday()) % 7
first_trading_day = first_day.replace(day=1 + days_to_monday)
else:
first_trading_day = first_day
return now.date() == first_trading_day.date()
elif time_interval == 'quarterly':
# 每季度第一个月的第一个交易日
if now.month in [1, 4, 7, 10] and check_date == 'first_trading_day':
first_day = now.replace(day=1)
if first_day.weekday() >= 5:
days_to_monday = (7 - first_day.weekday()) % 7
first_trading_day = first_day.replace(day=1 + days_to_monday)
else:
first_trading_day = first_day
return now.date() == first_trading_day.date()
elif time_interval == 'semi_annual':
# 每半年（1月和7月）
if now.month in [1, 7] and check_date == 'first_trading_day':
first_day = now.replace(day=1)
if first_day.weekday() >= 5:
days_to_monday = (7 - first_day.weekday()) % 7
first_trading_day = first_day.replace(day=1 + days_to_monday)
else:
first_trading_day = first_day
return now.date() == first_trading_day.date()
elif time_interval == 'annual':
# 每年第一个交易日
if now.month == 1 and check_date == 'first_trading_day':
first_day = now.replace(day=1)
if first_day.weekday() >= 5:
days_to_monday = (7 - first_day.weekday()) % 7
first_trading_day = first_day.replace(day=1 + days_to_monday)
else:
first_trading_day = first_day
return now.date() == first_trading_day.date()
return False
def _check_threshold_trigger(self):
"""
检查阈值触发条件
"""
# 获取当前持仓和价格
current_positions = self.fetcher.get_current_positions(self.trader.api)
current_prices = {}
for asset_name, asset_info in self.config['ASSET_CONFIG'].items():
symbol = asset_info['symbol']
market = asset_info['market']
price = self.fetcher.get_realtime_price(symbol, market)
if price:
current_prices[symbol] = price
if not current_positions or not current_prices:
self.logger.warning("无法获取持仓或价格数据，阈值检查失败")
return False
# 检查是否需要调仓
need_rebalance, reason = self.calculator.check_rebalance_needed(
current_positions, current_prices, self.config['TRADING_PARAMS']['total_capital']
)
return need_rebalance
def run_rebalance(self):
"""
执行一次完整的调仓流程
"""
self.logger.info("开始调仓流程...")
# 1. 检查交易时间
if not self.fetcher.check_trading_time():
self.logger.warning("当前非交易时间，跳过执行")
return False
# 2. 获取当前持仓
self.logger.info("步骤1: 获取当前持仓")
current_positions = self.fetcher.get_current_positions(self.trader.api)
if not current_positions:
self.logger.warning("当前无持仓，将进行初始建仓")
# 初始建仓逻辑
current_positions = {}
# 3. 获取当前价格
self.logger.info("步骤2: 获取当前价格")
current_prices = {}
for asset_name, asset_info in self.config['ASSET_CONFIG'].items():
symbol = asset_info['symbol']
market = asset_info['market']
price = self.fetcher.get_realtime_price(symbol, market)
if price:
current_prices[symbol] = price
else:
self.logger.error(f"无法获取{symbol}价格，终止调仓")
return False
# 4. 计算目标持仓
self.logger.info("步骤3: 计算目标持仓")
total_capital = self.config['TRADING_PARAMS']['total_capital']
target_shares = self.calculator.calculate_target_shares(current_prices, total_capital)
# 5. 计算交易指令
self.logger.info("步骤4: 计算交易指令")
trades = self.calculator.calculate_rebalance_trades(current_positions, target_shares)
if not trades:
self.logger.info("无需调仓，当前持仓已达标")
return True
# 6. 计算交易成本
cost_info = self.calculator.calculate_total_cost(trades)
self.logger.info(f"预估交易成本: {cost_info['total_cost']:.2f}元 (佣金: {cost_info['commission']:.2f}元，滑点: {cost_info['slippage']:.2f}元)")
# 7. 确认执行
if not self.simulate:
confirm = input(f"确认执行{len(trades)}笔交易？(y/n): ")
if confirm.lower() != 'y':
self.logger.info("用户取消交易")
return False
# 8. 执行交易
self.logger.info("步骤5: 执行交易")
results = self.trader.execute_rebalance(trades, simulate=self.simulate, delay=1)
# 9. 记录结果
self.logger.info("步骤6: 记录结果")
success_count = sum(1 for r in results if r['status'] in ['success', 'simulated'])
self.logger.info(f"调仓完成: 成功 {success_count}/{len(trades)} 笔")
return True
def run_continuous(self, interval=3600):
"""
连续运行模式（守护进程）
"""
self.logger.info(f"进入连续运行模式，检查间隔: {interval}秒")
try:
while True:
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
self.logger.info(f"--- {now} 开始检查 ---")
# 检查触发条件
if self.check_trigger_condition():
self.logger.info("触发条件满足，执行调仓")
self.run_rebalance()
else:
self.logger.info("触发条件不满足，跳过本次")
# 等待下次检查
self.logger.info(f"等待 {interval} 秒后下次检查...")
time.sleep(interval)
except KeyboardInterrupt:
self.logger.info("用户中断，系统停止")
except Exception as e:
self.logger.error(f"系统异常: {e}")
raise
def main():
"""
主函数
"""
# 选择运行模式
if len(sys.argv) > 1:
mode = sys.argv[1]
else:
mode = input("选择模式:\n1. 单次运行\n2. 连续运行\n请输入(1/2): ")
# 是否模拟
simulate = True # 默认模拟模式，安全第一
if len(sys.argv) > 2:
if sys.argv[2] == 'real':
simulate = False
else:
real_mode = input("是否实盘模式？(y/n, 默认n): ")
if real_mode.lower() == 'y':
simulate = False
# 创建系统实例
config = {
'STRATEGY_CONFIG': STRATEGY_CONFIG,
'ASSET_CONFIG': ASSET_CONFIG,
'TRADING_PARAMS': TRADING_PARAMS,
'ACCOUNT_INFO': ACCOUNT_INFO,
'LOG_CONFIG': LOG_CONFIG
}
system = AutoRebalanceSystem(config, simulate=simulate)
# 运行
if mode == '1':
# 单次运行
system.run_rebalance()
elif mode == '2':
# 连续运行
interval = 3600 # 每小时检查一次
system.run_continuous(interval=interval)
else:
print("无效模式")
return
if __name__ == "__main__":
main()
第五部分：实战操作指南
5.1 初次部署步骤
步骤1：环境准备
# 创建虚拟环境
python -m venv venv
source venv/bin/activate # Linux/Mac
# venv\Scripts\activate # Windows
# 安装依赖
pip install -r requirements.txt
步骤2：配置文件修改
- 修改
config.py中的账户信息 - 设置总资金量
- 选择交易标的和比例
步骤3：模拟测试
# 运行模拟模式
python main.py 1
# 或
python main.py 2 # 连续运行模拟
步骤4：实盘测试
- 先用小资金测试（如1万元）
- 连续观察3-6个月
- 确认无误后再加大资金
5.2 常见问题与解决方案
问题1：API连接失败
- 检查网络连接
- 确认API密钥正确
- 查看券商API状态
问题2：价格获取失败
- 切换数据源（AkShare/yfinance）
- 检查网络代理设置
- 增加重试逻辑
问题3：交易失败
- 检查账户资金是否充足
- 确认交易时间
- 检查最小交易单位
问题4：调仓过于频繁
- 调整触发阈值（如从5%改为10%）
- 改用时间触发模式
- 增加交易成本计算，过滤小额调仓
5.3 性能优化建议
1. 数据缓存
# 在data_fetcher.py中添加缓存
import pickle
import os
class DataFetcher:
def __init__(self):
self.cache_file = 'price_cache.pkl'
self.cache_timeout = 300 # 5分钟
def get_cached_price(self, symbol):
if os.path.exists(self.cache_file):
with open(self.cache_file, 'rb') as f:
cache = pickle.load(f)
if symbol in cache:
timestamp, price = cache[symbol]
if time.time() - timestamp < self.cache_timeout:
return price
return None
def save_cached_price(self, symbol, price):
cache = {}
if os.path.exists(self.cache_file):
with open(self.cache_file, 'rb') as f:
cache = pickle.load(f)
cache[symbol] = (time.time(), price)
with open(self.cache_file, 'wb') as f:
pickle.dump(cache, f)
2. 异步处理
import asyncio
import aiohttp
async def fetch_price_async(symbol):
async with aiohttp.ClientSession() as session:
# 异步获取价格
pass
3. 错误恢复机制
def run_with_retry(max_retries=3):
def decorator(func):
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == max_retries - 1:
raise
time.sleep(2 ** attempt) # 指数退避
return None
return wrapper
return decorator
5.4 监控与报警
添加邮件/短信通知
import smtplib
from email.mime.text import MIMEText
def send_alert(message):
"""
发送报警邮件
"""
try:
# 配置SMTP
smtp_server = "smtp.gmail.com"
smtp_port = 587
sender_email = "your_email@gmail.com"
sender_password = "your_password"
receiver_email = "receiver@gmail.com"
msg = MIMEText(message)
msg['Subject'] = "自动调仓系统报警"
msg['From'] = sender_email
msg['To'] = receiver_email
with smtplib.SMTP(smtp_server, smtp_port) as server:
server.starttls()
server.login(sender_email, sender_password)
server.send_message(msg)
logger.info("报警邮件发送成功")
except Exception as e:
logger.error(f"邮件发送失败: {e}")
第六部分：进阶策略与扩展
6.1 动态资产配置
基于估值的动态调整
def calculate_valuation_ratio(symbol, period="5y"):
"""
计算估值指标（如PE分位数）
"""
from data_fetcher import DataFetcher
fetcher = DataFetcher()
hist = fetcher.get_historical_data(symbol, period)
if hist is None or hist.empty:
return None
# 计算PE（简化版，实际需要基本面数据）
# 这里用市净率PB代替
current_price = hist['Close'].iloc[-1]
# 假设净资产为历史最高价的某个比例（实际需要真实数据）
max_price = hist['High'].max()
pb = current_price / max_price
# 计算当前PB在历史中的分位数
pb_history = hist['Close'] / max_price
percentile = (pb_history < pb).sum() / len(pb_history)
return percentile
def dynamic_allocation(current_prices, base_allocation):
"""
根据估值动态调整比例
"""
new_allocation = base_allocation.copy()
# 获取股票估值分位数
stock_pe = calculate_valuation_ratio('510300')
if stock_pe is not None:
if stock_pe > 0.8: # 估值过高
# 减少股票比例，增加债券比例
new_allocation['股票'] *= 0.9
new_allocation['债券'] *= 1.1
elif stock_pe < 0.2: # 估值过低
# 增加股票比例，减少债券比例
new_allocation['股票'] *= 1.1
new_allocation['债券'] *= 0.9
# 重新归一化
total = sum(new_allocation.values())
for key in new_allocation:
new_allocation[key] /= total
return new_allocation
6.2 多账户管理
支持多个账户同时管理
class MultiAccountManager:
def __init__(self, accounts_config):
self.accounts = {}
for acc_name, acc_config in accounts_config.items():
self.accounts[acc_name] = Trader(acc_config)
def run_all(self):
results = {}
for name, trader in self.accounts.items():
try:
# 执行调仓逻辑
results[name] = "成功"
except Exception as e:
results[name] = f"失败: {e}"
return results
6.3 回测系统
简单的回测框架
def backtest_strategy(strategy_func, data, initial_capital=100000):
"""
回测框架
"""
capital = initial_capital
positions = {}
trades = []
for date, prices in data.iterrows():
# 调用策略函数
target_allocation = strategy_func(prices)
# 计算调仓
# ...（类似主程序逻辑）
# 记录每日价值
# ...
# 计算收益指标
total_return = (capital - initial_capital) / initial_capital
annual_return = (1 + total_return) ** (252 / len(data)) - 1
return {
'total_return': total_return,
'annual_return': annual_return,
'trades': trades
}
第七部分：风险管理与合规
7.1 风险控制措施
1. 单资产上限
# 在config.py中添加
MAX_SINGLE_ASSET = 0.5 # 单资产不超过50%
2. 最大回撤控制
def check_max_drawdown(current_value, peak_value):
drawdown = (peak_value - current_value) / peak_value
if drawdown > 0.2: # 回撤超过20%
return True
return False
3. 交易频率限制
class TradeLimiter:
def __init__(self, max_trades_per_day=5):
self.max_trades = max_trades_per_day
self.trade_count = 0
self.last_reset = datetime.now().date()
def can_trade(self):
today = datetime.now().date()
if today != self.last_reset:
self.trade_count = 0
self.last_reset = today
if self.trade_count < self.max_trades:
self.trade_count += 1
return True
return False
7.2 合规注意事项
1. 了解当地法规
- 中国：T+1交易制度，涨跌停限制
- 美国：Pattern Day Trader规则
- 欧洲：MiFID II规定
2. 税务考虑
- 交易费用可抵税
- 资本利得税
- 持有期限影响税率
3. 账户安全
- API密钥加密存储
- 使用环境变量
- 定期更换密钥
第八部分：实战案例分析
8.1 案例：2020年疫情市场
市场背景：
- 2020年2-3月：全球股市暴跌
- 2020年4-12月：强劲反弹
60/40组合表现：
时间 股票价值 债券价值 总资产 操作
2020-01-01 60,000 40,000 100,000 -
2020-03-23 36,000 42,000 78,000 无需调仓（偏离4.6%）
2020-04-01 42,000 41,500 83,500 买入股票，卖出债券
2020-12-31 85,000 45,000 130,000 年度再平衡
关键教训：
- 市场暴跌时不要恐慌卖出
- 阈值触发比时间触发更有效
- 再平衡是”低买高卖”的纪律化执行
8.2 案例：2022年通胀高企
市场背景：
- 股债双杀
- 传统60/40组合失效
永久组合表现：
资产类别 配置比例 2022年表现
股票 25% -18%
长期国债 25% -12%
黄金 25% +5%
现金 25% +2%
组合整体 100% -5.5%
结论：永久组合在极端环境下更稳健
第九部分：总结与建议
9.1 核心要点回顾
- 策略选择：从简单开始，60/40组合最适合新手
- 工具选择：非程序员用券商条件单，程序员用Python
- 风险管理：模拟测试→小资金实盘→逐步加仓
- 持续优化：定期回顾，但不要过度优化
9.2 给不同投资者的建议
新手投资者：
- 从券商APP的智能投顾开始
- 使用定投+手动再平衡
- 学习基础概念后再尝试自动化
程序员/技术爱好者：
- 从模拟盘开始完整实现
- 优先考虑数据稳定性和API可靠性
- 建立完善的日志和监控
专业投资者：
- 考虑多策略组合
- 加入动态调整因子
- 建立风险预算模型
9.3 最终建议
懒人策略的真谛：
- 不是完全不管，而是”聪明地偷懒”
- 设置好规则，让系统替你执行
- 保持学习，但减少日常操作
记住黄金法则：
“Time in the market beats timing the market.” （在市场中待的时间比择时更重要）
通过自动调仓系统，你将真正做到”设置一次，长期受益”，让时间成为你的朋友而非敌人。
附录：完整项目文件清单
auto_rebalance/
├── config.py # 配置文件
├── data_fetcher.py # 数据获取
├── calculator.py # 计算逻辑
├── trader.py # 交易执行
├── logger.py # 日志系统
├── main.py # 主程序
├── requirements.txt # 依赖列表
├── auto_rebalance.log # 运行日志
├── price_cache.pkl # 价格缓存（可选）
└── README.md # 项目说明
requirements.txt：
pandas>=1.3.0
numpy>=1.20.0
yfinance>=0.1.70
akshare>=1.0.0
matplotlib>=3.4.0
运行命令：
# 模拟单次运行
python main.py 1
# 模拟连续运行
python main.py 2
# 实盘运行（警告：需要充分测试）
python main.py 1 real
python main.py 2 real
免责声明： 本文提供的代码和策略仅供学习参考。实际投资有风险，使用前请充分测试，并根据自身情况调整。建议先用模拟盘运行3-6个月，确认系统稳定后再考虑实盘。作者不对任何投资损失负责。