2051 北上资金持股比选股策略（北向港资外资）.py

# 克隆自聚宽文章：https://www.joinquant.com/post/29535
# 标题：北上资金持股比选股策略（北向/港资/外资）
# 作者：逆熵者

# 导入函数库
import pandas as pd
from jqdata import *

# 初始化函数
def initialize(context):
    # 设置基准
    g.benchmark = '000300.XSHG'
    set_benchmark(g.benchmark)
    # 使用真实价格并避免未来数据
    set_option('use_real_price', True)
    set_option("avoid_future_data", True)

# 为方便修改将变量置于此函数中
def after_code_changed(context):
    # 过滤掉order系列API产生的比error级别低的log
    log.set_level('order', 'error')
    # 股票类每笔交易时的手续费是：买入时佣金万分之三，卖出时佣金万分之三加千分之一印花税, 每笔交易佣金最低扣5块钱
    set_order_cost(OrderCost(close_tax=0.001, open_commission=0.00025, close_commission=0.00025, min_commission=5), type='stock')
    # set_order_cost(OrderCost(close_tax=0.0, open_commission=0.0, close_commission=0.0, min_commission=0), type='stock')
    # 设置基础股票池
    g.universe_index = '000902.XSHG'
    # 最大持股数量
    g.max_hold_stocknum = 10
    # 可以继续持股的排名
    g.check_out_ranking = 30
    # 个股最大最小仓位比例限制
    g.security_max_proportion = 0.20
    g.security_min_proportion = 0.05
    # 卖出后10交易日内不再买入
    g.n_tradeday_not_buy = 10
    # 初始化港资持股比例排名和已卖出股票列表
    g.prev_hk_hold_df = pd.DataFrame({})
    g.selled_security_dict = {}
    # run_weekly(main_func, 3, time='open', reference_security=g.benchmark)
    run_daily(main_func, time='open', reference_security=g.benchmark)
    run_daily(selled_security_list_count, time='after_close', reference_security=g.benchmark)

# 主函数
def main_func(context):
    prev_date = context.previous_date
    stock_list = get_index_stocks(g.universe_index)
    hk_hold_df = get_hk_hold_ratio(stock_list, end_date=prev_date, start_date=prev_date, sorted_by_circulation=False)
    # 查询结果为空时沿用上一次的有效查询结果
    if len(hk_hold_df) == 0:
        log.info('港资查询失败，沿用上次查询结果')
        hk_hold_df = g.prev_hk_hold_df
    else:
        g.prev_hk_hold_df = hk_hold_df
    hold_lists = list(hk_hold_df['code'])
    # 过滤ST停牌退市
    hold_lists = st_filter(context, hold_lists)
    hold_lists = paused_filter(context, hold_lists)
    hold_lists = delisted_filter(context, hold_lists)
    hold_lists = hold_lists[:g.check_out_ranking]
    trade(context, hold_lists, g.max_hold_stocknum, g.security_max_proportion, g.security_min_proportion)

## 卖出股票日期计数
def selled_security_list_count(context):
    selled_num = len(g.selled_security_dict)
    if selled_num > 0:
        log.info('累积清仓标的数：%s' % selled_num)
        for stock in g.selled_security_dict.keys():
            g.selled_security_dict[stock] += 1


## 过滤同一标的继上次卖出N天不再买入
def filter_n_tradeday_not_buy(security, n=0):
    try:
        if (security in g.selled_security_dict.keys()) and (g.selled_security_dict[security] < n): log.info('%s %s交易日前曾卖出，不符合买入要求(%s)' % (security, g.selled_security_dict[security], n)) return False return True except: return True ## 卖出股票加入dict def selled_security_list_dict(context, security_list): selled_sl = [s for s in security_list if s not in context.portfolio.positions.keys()] if len(selled_sl)>0:
        for stock in selled_sl:
            g.selled_security_dict[stock] = 0

## 过滤停牌股票
def paused_filter(context, security_list):
    current_data = get_current_data()
    security_list = [stock for stock in security_list if not current_data[stock].paused]
    return security_list

## 过滤退市股票
def delisted_filter(context, security_list):
    current_data = get_current_data()
    security_list = [stock for stock in security_list if not (('退' in current_data[stock].name) or ('*' in current_data[stock].name))]
    return security_list

## 过滤ST股票
def st_filter(context, security_list):
    current_data = get_current_data()
    security_list = [stock for stock in security_list if not current_data[stock].is_st]
    return security_list

# 过滤涨停股票
def high_limit_filter(context, security_list):
    current_data = get_current_data()
    security_list = [stock for stock in security_list if not (current_data[stock].day_open >= current_data[stock].high_limit)]
    return security_list


解锁查看函数：

• 交易函数  def trade()
• 获取港资持股比例 def get_hk_hold_ratio()