# 导入函数库
from jqdata import *
import numpy as np
# 初始化函数，设定基准等等
def initialize(context):
# 设定上证50作为基准
set_benchmark('600016.XSHG')
# 开启动态复权模式(真实价格)
set_option('use_real_price', True)
# 输出内容到日志 log.info()
log.info('初始函数开始运行且全局只运行一次')
# 过滤掉order系列API产生的比error级别低的log
log.set_level('order', 'error')
### 股票相关设定 ###
# 股票类每笔交易时的手续费是：买入时佣金万分之三，卖出时佣金万分之三加千分之一印花税, 每笔交易佣金最低扣5块钱
set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
# 策略参数
g.security = '600016.XSHG'
g.ma_short = 5 # 短期均线
g.ma_long = 20 # 长期均线
g.rsi_period = 14 # RSI参数
g.oversold_threshold = 30 # RSI超卖阈值
g.overbought_threshold = 70 # RSI超买阈值
g.max_position_pct = 0.3 # 最大仓位比例
g.stop_loss_pct = -8 # 止损点
g.trailing_stop_pct = 5 # 追踪止损点
g.take_profit_pct = 25 # 止盈点
# 设置每日运行函数
run_daily(before_market_open, time='before_open')
run_daily(market_open, time='every_bar')
# 计算RSI指标
def calculate_rsi(prices, period=14):
deltas = np.diff(prices)
seed = deltas[:period+1]
up = seed[seed >= 0].sum()/period
down = -seed[seed < 0].sum()/period
rs = up/down if down != 0 else 0
rsi = np.zeros_like(prices)
rsi[:period] = 100. - 100./(1. + rs)
for i in range(period, len(prices)):
delta = deltas[i-1]
if delta > 0:
upval = delta
downval = 0.
else:
upval = 0.
downval = -delta
up = (up * (period - 1) + upval) / period
down = (down * (period - 1) + downval) / period
rs = up/down if down != 0 else 0
rsi[i] = 100. - 100./(1. + rs)
return rsi
## 开盘前运行函数
def before_market_open(context):
# 输出运行时间
log.info('函数运行时间(before_market_open)：'+str(context.current_dt.time()))
# 获取数据并计算指标
security = g.security
# 获取历史数据
prices = get_bars(security, count=g.ma_long+10, unit='1d', fields=['close'])
close_prices = prices['close']
# 计算均线
g.ma_short_value = close_prices[-g.ma_short:].mean()
g.ma_long_value = close_prices[-g.ma_long:].mean()
# 计算RSI
g.rsi_value = calculate_rsi(close_prices, g.rsi_period)[-1]
log.info('短期均线: {}, 长期均线: {}, RSI: {}'.format(
round(g.ma_short_value, 2),
round(g.ma_long_value, 2),
round(g.rsi_value, 2)
))
## 开盘时运行函数
def market_open(context):
log.info('函数运行时间(market_open):'+str(context.current_dt.time()))
security = g.security
# 获取当前价格
current_price = get_current_data()[security].last_price
log.info('当前价格：'+str(current_price))
# 获取当前持仓
position = context.portfolio.positions.get(security, None)
position_amt = 0 if position is None else position.total_amount
# 计算仓位比例
total_value = context.portfolio.total_value
current_position_pct = 0 if position is None else position.value / total_value
available_cash = context.portfolio.available_cash
log.info('当前持仓比例: {}%, 当前现金: {}'.format(
round(current_position_pct * 100, 2),
round(available_cash, 2)
))
# 买入条件:
# 1. 短期均线上穿长期均线
# 2. RSI处于超卖区域反弹
# 3. 当前仓位小于目标最大仓位
ma_crossover = g.ma_short_value > g.ma_long_value
rsi_buy_signal = g.rsi_value > g.oversold_threshold and g.rsi_value < 50
if ma_crossover and rsi_buy_signal and current_position_pct < g.max_position_pct:
# 计算买入金额 (Target position - current position)
target_position_pct = g.max_position_pct
target_value = total_value * target_position_pct
current_value = 0 if position is None else position.value
buy_value = target_value - current_value
if buy_value > 5000: # 确保买入金额有意义
log.info('满足买入条件，买入金额: {}'.format(round(buy_value, 2)))
order_value(security, buy_value)
# 卖出逻辑
if position is not None and position.total_amount > 0:
# 计算收益率
earn_ratio = (position.price / position.avg_cost - 1) * 100
log.info('当前收益：{}%'.format(round(earn_ratio, 2)))
# 1. 止损: 亏损达到止损点
# 2. 止盈: 盈利达到止盈点
# 3. 技术指标: RSI进入超买区域
stop_loss_triggered = earn_ratio < = g.stop_loss_pct
take_profit_triggered = earn_ratio >= g.take_profit_pct
rsi_sell_signal = g.rsi_value >= g.overbought_threshold
ma_sell_signal = g.ma_short_value < g.ma_long_value
# 追踪止损: 记录最高收益率，回撤超过设定比例时卖出
if not hasattr(g, 'highest_earn_ratio') or earn_ratio > g.highest_earn_ratio:
g.highest_earn_ratio = earn_ratio
trailing_stop_triggered = (g.highest_earn_ratio - earn_ratio) >= g.trailing_stop_pct and g.highest_earn_ratio > 10
if stop_loss_triggered:
log.info('触发止损，卖出全部')
order_target(security, 0)
elif take_profit_triggered:
log.info('触发止盈，卖出全部')
order_target(security, 0)
elif trailing_stop_triggered:
log.info('触发追踪止损，卖出全部')
order_target(security, 0)
elif rsi_sell_signal and ma_sell_signal:
log.info('技术指标显示超买，卖出一半')
order_target_value(security, position.value * 0.5)
2025-04-13
雪球
