"""
|
掘金
|
"""
|
|
from __future__ import print_function, absolute_import
|
|
import datetime
|
import json
|
import logging
|
import time as t
|
|
import numpy
|
import schedule
|
|
import gm.api as gmapi
|
|
import big_money_num_manager
|
import client_manager
|
import code_volumn_manager
|
import constant
|
import global_data_loader
|
import global_util
|
import gpcode_first_screen_manager
|
import gpcode_manager
|
import threading
|
|
import server
|
import tool
|
|
from db import redis_manager
|
import authority
|
import decimal
|
|
from trade import trade_gui, l2_trade_util, trade_manager
|
from l2.cancel_buy_strategy import L2LimitUpSellStatisticUtil
|
from log import logger_juejin_tick, logger_system
|
from trade.trade_data_manager import CodeActualPriceProcessor
|
from trade.trade_queue_manager import JueJinBuy1VolumnManager
|
|
redisManager = redis_manager.RedisManager(0)
|
__jueJinBuy1VolumnManager = JueJinBuy1VolumnManager()
|
__actualPriceProcessor = CodeActualPriceProcessor()
|
|
|
# 设置账户信息
|
def setAccountInfo(accountId, strategyId, token):
|
redis = redisManager.getRedis()
|
redis.set("juejin-account-id", accountId)
|
redis.set("juejin-strategy-id", strategyId)
|
redis.set("juejin-token", token)
|
|
|
def getAccountInfo():
|
redis = redisManager.getRedis()
|
account_id = redis.get("juejin-account-id")
|
strategy_id = redis.get("juejin-strategy-id")
|
token = redis.get("juejin-token")
|
return account_id, strategy_id, token
|
|
|
def init_data():
|
# 删除所有的涨停卖数据
|
L2LimitUpSellStatisticUtil.clear()
|
# 重置所有的大单数据
|
big_money_num_manager.reset_all()
|
# 清除水下捞数据
|
__actualPriceProcessor.clear_under_water_data()
|
# 载入行业股票代码
|
global_data_loader.load_industry()
|
# 载入代码自由流通市值
|
global_data_loader.load_zyltgb()
|
# 载入量
|
global_data_loader.load_volumn()
|
|
# 9点25之前删除所有代码
|
if tool.trade_time_sub(tool.get_now_time_str(), "09:25:00") <= 0:
|
# 删除L2监听代码
|
gpcode_manager.clear_listen_codes()
|
# 删除首板代码
|
gpcode_manager.clear_first_codes()
|
# 删除首板未筛选代码
|
gpcode_first_screen_manager.clear_first_no_screen_codes()
|
# 删除禁止代码
|
l2_trade_util.init_forbidden_trade_codes()
|
# 清空白名单
|
l2_trade_util.WhiteListCodeManager.clear()
|
# 清空想要买
|
gpcode_manager.WantBuyCodesManager.clear()
|
# 清空分数禁止代码
|
trade_manager.ForbiddenBuyCodeByScoreManager.clear()
|
|
|
# 每日初始化
|
def everyday_init():
|
# 交易時間不能做初始化
|
if not tool.is_init_time():
|
raise Exception("交易时间不能初始化")
|
|
codes = gpcode_manager.get_gp_list()
|
logger_system.info("每日初始化")
|
|
# 今日实时涨停
|
global_data_loader.add_limit_up_codes([], True)
|
# 主要获取收盘价
|
__get_latest_info(None)
|
# 获取60天最大量与昨日量
|
global_util.today_volumn.clear()
|
global_util.max60_volumn.clear()
|
global_util.yesterday_volumn.clear()
|
volumn_dict = get_volumns(codes)
|
for key in volumn_dict:
|
code_volumn_manager.set_histry_volumn(key, volumn_dict[key][0], volumn_dict[key][1])
|
# 清除大单数据
|
global_util.big_money_num.clear()
|
# 初始化大单数据
|
for code in codes:
|
big_money_num_manager.add_num(code, 0)
|
big_money_num_manager.expire(code)
|
# 清除涨停时间
|
global_util.limit_up_time.clear()
|
init_data()
|
# 初始化同花顺主站
|
l2_clients = client_manager.getValidL2Clients()
|
for client in l2_clients:
|
try:
|
server.repair_ths_main_site(client)
|
except Exception as e:
|
pass
|
|
|
def __run_schedule():
|
while True:
|
schedule.run_pending()
|
|
|
def init(context):
|
# gmapi.subscribe(symbols="SZSE.002529", frequency="1d", count=30)
|
|
# 订阅浦发银行, bar频率为一天和一分钟
|
# 订阅订阅多个频率的数据，可多次调用subscribe
|
# 获取需要监听的股票
|
init_data()
|
logger_system.info("掘金初始化")
|
schedule.every().day.at("09:15:00").do(everyday_init)
|
t1 = threading.Thread(target=lambda: __run_schedule())
|
# 后台运行
|
t1.setDaemon(True)
|
t1.start()
|
|
# 多个时间点获取收盘价
|
gmapi.schedule(schedule_func=__get_latest_info, date_rule='1d', time_rule='08:30:00')
|
gmapi.schedule(schedule_func=__get_latest_info, date_rule='1d', time_rule='08:50:00')
|
gmapi.schedule(schedule_func=__get_latest_info, date_rule='1d', time_rule='09:28:00')
|
gmapi.schedule(schedule_func=__get_current_info, date_rule='1d', time_rule='09:25:00')
|
gmapi.schedule(schedule_func=__get_current_info, date_rule='1d', time_rule='09:29:00')
|
re_subscribe_tick()
|
# re_subscribe_bar()
|
|
# 初始化内容
|
clients = authority.get_l2_clients()
|
for client in clients:
|
for i in range(0, constant.L2_CODE_COUNT_PER_DEVICE):
|
gpcode_manager.init_listen_code_by_pos(client, i)
|
|
|
def __get_latest_info(context):
|
# 初始化内容
|
clients = authority.get_l2_clients()
|
for c in clients:
|
for i in range(0, constant.L2_CODE_COUNT_PER_DEVICE):
|
gpcode_manager.init_listen_code_by_pos(int(c), i)
|
codes = gpcode_manager.get_gp_list()
|
result = JueJinManager.get_gp_latest_info(codes)
|
for item in result:
|
sec_level = item['sec_level']
|
symbol = item['symbol']
|
symbol = symbol.split(".")[1]
|
pre_close = tool.to_price(decimal.Decimal(str(item['pre_close'])))
|
if sec_level == 1:
|
if symbol in codes:
|
gpcode_manager.set_price_pre(symbol, pre_close)
|
else:
|
gpcode_manager.rm_gp(symbol)
|
|
|
# 获取最新的信息
|
def __get_current_info():
|
data = gpcode_manager.get_gp_list()
|
results = JueJinManager.get_gp_current_info(data)
|
logger_juejin_tick.debug("定时获取：{}", results)
|
for result in results:
|
price = result["price"]
|
symbol = result['symbol']
|
# 保存最新价
|
symbol = symbol.split(".")[1]
|
|
|
# 设置收盘价
|
def re_set_price_pre(code):
|
codes = [code]
|
re_set_price_pres(codes)
|
|
|
def re_set_price_pres(codes, force=False):
|
result = JueJinManager.get_gp_latest_info(codes)
|
for item in result:
|
symbol = item['symbol']
|
symbol = symbol.split(".")[1]
|
pre_close = tool.to_price(decimal.Decimal(str(item['pre_close'])))
|
gpcode_manager.set_price_pre(symbol, pre_close, force)
|
|
|
__prices_now = {}
|
|
|
def on_tick(context, tick):
|
# print(tick["created_at"])
|
relative_timestamp = t.time() % (24 * 60 * 60) + 8 * 60 * 60
|
# 9点20-15:05接受数据
|
start1 = 60 * 60 * 9 + 31 * 60
|
end1 = 60 * 60 * 11 + 35 * 60
|
start2 = 60 * 60 * 12 + 50 * 60
|
end2 = 60 * 60 * 15 + 5 * 60
|
# TODO 测试
|
if (start1 < relative_timestamp < end1 or start2 < relative_timestamp < end2) or constant.TEST:
|
symbol = tick['symbol']
|
price = tick['price']
|
# print(symbol,price)
|
# 价格有变化的时候才发送操作
|
# if symbol in __prices_now and __prices_now[symbol] == price:
|
# return
|
|
# 保存最新价
|
symbol = symbol.split(".")[1]
|
JueJinManager.add_listen_code(symbol)
|
time_ = tick["created_at"].strftime("%H:%M:%S")
|
data_ = (symbol, time_, tick["quotes"][0]["bid_v"], tick["quotes"][0]["bid_p"])
|
logger_juejin_tick.info("买1量 {},{},{},{}", data_[1], data_[0], data_[2],
|
data_[3])
|
# 暂时不采用
|
# need_sync = __jueJinBuy1VolumnManager.save(data_[0], data_[1], data_[2], data_[3])
|
# if need_sync:
|
# # 同步数据
|
# L2LimitUpMoneyStatisticUtil.verify_num(data_[0], data_[2], data_[1])
|
|
# print(tick["created_at"],tick["quotes"][0]["bid_v"])
|
__prices_now[symbol] = price
|
|
def on_bar(context, bars):
|
print("on_bar", bars)
|
|
|
def re_subscribe_tick():
|
gpcode_list = gpcode_manager.get_gp_list_with_prefix()
|
codes_str = ",".join(gpcode_list)
|
print("订阅：", codes_str)
|
# TODO
|
gmapi.subscribe(symbols=codes_str, frequency='tick', count=2, wait_group=False, wait_group_timeout='10s',
|
unsubscribe_previous=True)
|
|
|
def re_subscribe_bar():
|
gpcode_list = gpcode_manager.get_gp_list_with_prefix()
|
codes_str = ",".join(gpcode_list)
|
print("订阅：", codes_str)
|
gmapi.subscribe(symbols=codes_str, frequency='60s', count=1,
|
unsubscribe_previous=True)
|
|
|
def recieve_msg(pipe):
|
while True:
|
value = pipe.recv()
|
print("跨进程通信：", value)
|
jsonValue = json.loads(value)
|
action = jsonValue["type"]
|
if action == 'resub':
|
re_subscribe_tick()
|
elif action == 'accpt_price':
|
try:
|
datas = jsonValue["data"]
|
# for data in datas:
|
# accpt_price(data["code"], float(data["price"]))
|
except Exception as e:
|
print(str(e))
|
|
|
class JueJinManager:
|
def __init__(self, pipe):
|
self.pipe = pipe
|
# 开启线程接受数据
|
t1 = threading.Thread(target=lambda: recieve_msg(pipe))
|
t1.setDaemon(True)
|
t1.start()
|
|
@classmethod
|
def get_gp_latest_info(cls, codes):
|
account_id, s_id, token = getAccountInfo()
|
symbols = gpcode_manager.get_gp_list_with_prefix(codes)
|
gmapi.set_token(token)
|
data = gmapi.get_instruments(symbols=",".join(symbols))
|
print(data)
|
return data
|
|
@classmethod
|
def get_now_price(cls, codes):
|
data = JueJinManager.get_gp_current_info(codes)
|
prices = []
|
for item in data:
|
code = item["symbol"].split(".")[1]
|
price = item["price"]
|
prices.append((code, price))
|
return prices
|
|
# 获取代码的涨幅
|
@classmethod
|
def get_codes_limit_rate(cls, codes):
|
datas = JueJinManager.get_gp_latest_info(codes)
|
pre_price_dict = {}
|
for data in datas:
|
code = data["sec_id"]
|
pre_close = tool.to_price(decimal.Decimal(str(data['pre_close'])))
|
pre_price_dict[code] = pre_close
|
|
now_prices = JueJinManager.get_now_price(codes)
|
f_results = []
|
for data in now_prices:
|
code = data[0]
|
price = data[1]
|
pre_price = float(pre_price_dict.get(code))
|
rate = round((price - pre_price) * 100 / pre_price, 2)
|
f_results.append((code, rate))
|
f_results.sort(key=lambda tup: tup[1])
|
f_results.reverse()
|
return f_results
|
|
@classmethod
|
def get_history_tick_n(cls, code, count, fields=None):
|
account_id, s_id, token = getAccountInfo()
|
symbols = gpcode_manager.get_gp_list_with_prefix([code])
|
gmapi.set_token(token)
|
results = gmapi.history_n(symbol=symbols[0], frequency="1d", count=count, fields=fields)
|
return results
|
|
@classmethod
|
def get_lowest_price_rate(cls, code, count):
|
datas = cls.get_history_tick_n(code, count)
|
low_price = datas[0]["close"]
|
for data in datas:
|
if low_price > data["close"]:
|
low_price = data["close"]
|
return (datas[-1]["close"] - low_price) / low_price
|
|
@classmethod
|
def get_gp_current_info(cls, codes):
|
account_id, s_id, token = getAccountInfo()
|
symbols = gpcode_manager.get_gp_list_with_prefix(codes)
|
gmapi.set_token(token)
|
data = gmapi.current(symbols=",".join(symbols))
|
print(data)
|
return data
|
|
@classmethod
|
def get_gp_codes_names(cls, codes):
|
datas = cls.get_gp_latest_info(codes)
|
results = {}
|
for data in datas:
|
code = data["symbol"].split(".")[1]
|
code_name = data['sec_name']
|
results[code] = code_name
|
return results
|
|
def start(self):
|
account_id, s_id, token = getAccountInfo()
|
gmapi.run(strategy_id=s_id,
|
filename='juejin.py',
|
# todo 回测模式
|
mode=gmapi.MODE_LIVE,
|
backtest_start_time='2022-08-18 09:25:00',
|
backtest_end_time='2022-08-18 10:30:00',
|
token=token)
|
|
def stop(self):
|
gmapi.stop()
|
|
@classmethod
|
def add_listen_code(cls, code):
|
redis = redisManager.getRedis()
|
redis.setex("juejin_listen_code-{}".format(code), 20, "1")
|
|
@classmethod
|
def get_listen_codes_lenth(cls):
|
redis = redisManager.getRedis()
|
keys = redis.keys("juejin_listen_code-*")
|
return len(keys)
|
|
# 返回指定日期的上个交易日
|
@classmethod
|
def get_previous_trading_date(cls, date):
|
account_id, s_id, token = getAccountInfo()
|
gmapi.set_token(token)
|
return gmapi.get_previous_trading_date("SHSE", date)
|
|
# 返回指定日期的下个交易日
|
@classmethod
|
def get_next_trading_date(cls, date):
|
account_id, s_id, token = getAccountInfo()
|
gmapi.set_token(token)
|
return gmapi.get_previous_trading_date("SHSE", date)
|
|
@classmethod
|
def get_trading_dates(cls, start_date, end_date):
|
account_id, s_id, token = getAccountInfo()
|
gmapi.set_token(token)
|
return gmapi.get_trading_dates("SHSE", start_date, end_date)
|
|
|
# 获取近90天的最大量与最近的量
|
def get_volumns(codes):
|
end = datetime.datetime.now()
|
# 获取近90天的历史数据
|
cha = datetime.timedelta(days=90)
|
start = end - cha
|
|
account_id, s_id, token = getAccountInfo()
|
gmapi.set_token(token)
|
gmapi.set_account_id(account_id)
|
results = gmapi.history(symbol=gpcode_manager.get_gp_list_with_prefix(codes), frequency="1d",
|
start_time="{:%Y-%m-%d}".format(start),
|
fields="symbol,volume,eob",
|
end_time="{:%Y-%m-%d}".format(end))
|
_fresult = {}
|
|
for result in results:
|
code = result["symbol"].split(".")[1]
|
volumn = int(result["volume"])
|
day = "{:%Y-%m-%d}".format(result["eob"])
|
|
if _fresult.get(code) is None:
|
_fresult[code] = (volumn, volumn)
|
|
if volumn > _fresult[code][0]:
|
_fresult[code] = (volumn, _fresult[code][1])
|
_fresult[code] = (_fresult[code][0], volumn)
|
return _fresult
|
|
|
# 获取近90天的最大量与最近的量
|
# 获取最近一次涨停/涨停下一个交易日的最大值
|
def get_volumns_by_code(code, count=60) -> object:
|
datas = JueJinManager.get_history_tick_n(code, count, "open,high,low,close,volume,pre_close,bob")
|
# 计算
|
datas.sort(key=lambda x: x["bob"], reverse=True)
|
return datas
|
|
|
# 解析最大量
|
def parse_max_volume(datas, is_new_top=False):
|
max_volume = 0
|
|
max_volume_date = None
|
if is_new_top:
|
# 如果是突破前高就取最大量
|
for item in datas:
|
if max_volume < item["volume"]:
|
max_volume = item["volume"]
|
max_volume_date = item["bob"]
|
else:
|
date = None
|
for i in range(len(datas)):
|
# 查询涨停
|
item = datas[i]
|
volume = item["volume"]
|
if max_volume < volume:
|
max_volume = volume
|
max_volume_date = item['bob']
|
# 是否有涨停
|
limit_up_price = float(gpcode_manager.get_limit_up_price_by_preprice(item["pre_close"]))
|
if abs(limit_up_price - item["high"]) < 0.01:
|
next_volume = 0
|
if i > 0:
|
next_volume = datas[i - 1]["volume"]
|
date = datas[i]["bob"]
|
if volume < next_volume:
|
volume = next_volume
|
date = datas[i - 1]["bob"]
|
return volume, volume, date.strftime("%Y-%m-%d")
|
return max_volume, max_volume, max_volume_date.strftime("%Y-%m-%d")
|
|
|
if __name__ == '__main__':
|
init_data()
|
