"""
|
该模块为全局化变量的模块
|
不要试图把所有数据都全局化了
|
"""
|
|
import datetime
|
import json
|
import logging
|
# 引入掘金API
|
from gm.api import *
|
|
|
# 全局日志配置 如果不想用就把 logging.ERROR 如果要打就=logging.INFO
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
|
logging.info("全局初始化数据 开始》》》")
|
|
# 设置账户ID
|
account_id = 'aaee2221-839c-11ee-a7cd-00163e022aa6'
|
# 设置token
|
token = '6c1dbe95191fb77cced9d805cb9c853805551ddb'
|
set_token("6c1dbe95191fb77cced9d805cb9c853805551ddb")
|
# 设置策略ID
|
strategy_id = '507bb60a-a919-11ee-b6ad-0ae0afd621cd'
|
|
# 全局化上下文对象context
|
context = None
|
|
# 今日时间初始化
|
now = datetime.datetime.now() # 获取本机时间
|
print(f"本机时间::{now}")
|
# 全局化 【今日日期 上一个交易日期 上上个交易日期 下一个交易日期】
|
today_date = now.strftime('%Y-%m-%d') # 获取今日日期参数
|
print(f"今日日期{today_date}")
|
pre_trading_day = get_previous_trading_date(exchange='SHSE', date=today_date) # 获取前一个交易日
|
print(f"上一个交易日{pre_trading_day}")
|
double_pre_trading_day = get_previous_trading_date(exchange='SHSE', date=pre_trading_day) # 获取上上一个交易日
|
print(f"上上一个交易日{double_pre_trading_day}")
|
next_trading_day = get_next_trading_date(exchange='SHSE', date=today_date) # 获取下一个交易日
|
print(f"下一个交易日{next_trading_day}")
|
|
# 获取A股市场（包含沪深两市）的股票列表跳过停牌，跳过ST 上交所 SHSE.600000 深交所 SZSE.000000 target = ['SHSE.603839', 'SZSE.002855']
|
all_stocks = get_instruments(exchanges='SZSE,SHSE', sec_types=1, fields='symbol', df=True, skip_suspended=True, skip_st=True)['symbol'].tolist()
|
print("A股所有代码数量:", len(all_stocks))
|
# 只要上证A股和深证A股
|
filtered_stocks = [stock for stock in all_stocks if stock.startswith('SHSE.60') or (stock.startswith('SZSE.00'))]
|
print(f"过滤后上证A股和深证A股数量:{len(filtered_stocks)}")
|
# 获取上证A股和深证A股 基本信息
|
instruments = get_instruments(symbols=filtered_stocks, exchanges=None, sec_types=None, names=None, fields="symbol,sec_name,pre_close", df=False)
|
# basic_info_symbols_dict = {} # 初始化A股代码字典
|
# 将获取到的上证A股和深证A股 基本信息 汇编为一个字典
|
basic_info_symbols_dict = {}
|
for i in instruments:
|
basic_info_symbols_dict[i["symbol"]] = i
|
# print(f"basic_info_symbols_dict ====={basic_info_symbols_dict}")
|
# basic_info_symbols_dict ===== 'SHSE.600611': {'symbol': 'SHSE.600611', 'sec_name': '大众交通', 'pre_close': 2.809999942779541},
|
# 将列表推导式移出循环 过滤出昨日收盘价在5-30元的股票
|
min_instruments = [stock for stock in instruments if 2 < stock['pre_close'] < 30]
|
# print(f"min_instruments 数量==={len(min_instruments)}")
|
# 使用 get 方法安全地获取 symbol 字段的值
|
min_stocks = [stock.get('symbol') for stock in min_instruments if stock.get('symbol') is not None]
|
print(f"min_stocks==={len(min_stocks)}")
|
|
# 读取已经获取到并存储在本地的目标范围的个股的板块概念
|
# 读取JSON文件并解析为字典
|
with open('local_storage_data/all_stocks_plate_dict.json', 'r', encoding='utf-8') as f:
|
json_data = f.read()
|
all_stocks_plate_dict = json.loads(json_data)
|
# print(f"all_stocks_plate_dict===={len(all_stocks_plate_dict)}个")
|
|
# 全局化涨停概念板块
|
limit_up_block_names = []
|
# 全局化板块强度下的个股强度
|
market_sift_plate_stock_dict = {}
|
# 为持仓代码创建一个初始集合，并全局化
|
position_symbols_set = set()
|
# 为当日可用持仓代码创建一个初始集合，并全局化
|
available_symbols_set = set()
|
# 为当日新增加持仓代码创建一个初始集合，并全局化
|
addition_position_symbols_set = set()
|
# 为K线属性指标字典初始化全局化
|
all_stocks_all_K_line_property_dict = {}
|
|
|
logging.info("全局初始化数据 完成《《《")
|
