直接下载数据文件即可导入 无需重新创建
gp.py
import pandas as pd
import numpy as np
import baostock as bs
import requests
from datetime import datetime, timedelta
import time
# import talib as ta
#使用quant()函数替换
class GP:
def __init__(self):
self.login()
def login(self):
bs.login()
def logout(self):
bs.logout()
def get_history(self,gp_type,code,end_date=False,start_date=False,days=14):
if not end_date:
end_date=datetime.now().strftime("%Y-%m-%d")
if not start_date:
start_date=(datetime.strptime(end_date, "%Y-%m-%d") - timedelta(days)).strftime("%Y-%m-%d")
rs = bs.query_history_k_data_plus(
code=gp_type+"."+str(code),
fields="code,date,open,high,low,close,preclose,volume,pctChg,turn,isST,adjustflag", # 要获取的字段（手动指定，Akshare 自动返回全字段）
start_date=start_date, # 日期格式：必须带横杠（Akshare 可无）
end_date=end_date,
frequency="d", # 周期：d=日线（和 Akshare 的 period="daily" 对应）
adjustflag="2" # 复权：2=前复权（对应 Akshare 的 adjust="qfq"）
)
# 3. 转为DataFrame（核心步骤）
df = rs.get_data()
df = df.replace("", 0) # 把所有空字符串变成 0
if df['preclose'].iloc[-1]=="":
df.drop(df.index[-1],inplace=True)
# 4. 数据类型转换（避免数值以字符串显示）
df = df.astype({
"code":str,
"open": float,
"high": float,
"low": float,
"close": float,
"preclose": float,
"volume": int,
"pctChg":float,
"turn":float,
"isST":int,
"adjustflag": int # 复权标识转整数
})
return df
#股票代码数据
def gp_code(self,code):
str_code=str(code)
if str_code[0] in ['0','1','2','3']:
gp_type='sz'
elif str_code[0] in ['5','6']:
gp_type='sh'
elif str_code[0] in ['4','8','9']:
gp_type='bj'
else:
return False
return [gp_type,str_code]
#当日股票
def get_today(self,gp_type,code):
str_code=str(code)
url="https://qt.gtimg.cn/q="
url=url+gp_type+str_code
headers = {
'User-Agent': "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/133.0.0.0 Safari/537.36"
}
res=requests.get(url,headers=headers)
if res.status_code==200:
if len(res.content)>30:
reb = res.content.decode("gbk")
reb = reb.replace("~~~", "~").replace("~~", "~")
vsx = reb.split('~')
return [
{
"code":gp_type+"."+str_code,
"date":datetime.now().strftime("%Y-%m-%d"),
"open": float(vsx[5]),
"high": float(vsx[32]),
"low": float(vsx[33]),
"close": float(vsx[3]),
"volume": float(vsx[6])*100,
"adjustflag": 2 # 复权标识转整数 统一baostock
},
{
"name":vsx[1],
'money':vsx[42]
}
]
return False
def gupiao(self,gp_type,gp_code,days=180):
#获得最近180天的交易 包括今天 解决baostock 没有今日数据
if "9:30"<=datetime.now().strftime('%H:%M')<"17:30" and datetime.now().weekday() in [0,1,2,3,4]:
ma=self.get_today(gp_type,gp_code)
if ma:
if int(ma[0]['volume']) != 0:
df=self.get_history(gp_type,gp_code,days=days)
df = pd.concat([df, pd.DataFrame([ma[0]])], ignore_index=True)
return df
return False
else:
print("user 17:30",gp_type,gp_code)
return self.get_history(gp_type,gp_code,days=days)
def quant(self,df):
#处理量化指标
#VOL MA
df['vol_ma5']=df["volume"].rolling(window=5).mean()
df['vol_ma10']=df["volume"].rolling(window=10).mean()
df['vol_ma20']=df["volume"].rolling(window=20).mean()
df['vol_ma60']=df["volume"].rolling(window=60).mean()
#MA
df['ma5']=df["close"].rolling(window=5).mean()
df['ma10']=df["close"].rolling(window=10).mean()
df['ma20']=df["close"].rolling(window=20).mean()
df['ma60']=df["close"].rolling(window=60).mean()
#ATR
df['hl']=df['high']-df['low']
df['hc']=abs(df['high']-df['close'].shift(1))
df['lc']=abs(df['low']-df['close'].shift(1))
df['tr']=df[['hl','hc','lc']].abs().max(axis=1) #axis=1行计算
df['atr'] = df['tr'].ewm(span=14, adjust=False).mean()
#RSI 14 6
df['delta']=df['close'].diff()
df['gain']=df['delta'].where(df['delta']>0,0)
df['loss']=-df['delta'].where(df['delta']<0,0)
df['avg_gain(14)']=df['gain'].rolling(14).mean()
df['avg_loss(14)']=df['loss'].rolling(14).mean()
df['avg_loss(14)'] = df['avg_loss(14)'].replace(0, 0.01)
df['rs_14']=df['avg_gain(14)']/df['avg_loss(14)']
df['rs_14'] = df['rs_14'].clip(0, 100) # 限制在0-100
df['rsi_14']=100-(100/(1+df['rs_14']))
df['avg_gain(6)']=df['gain'].rolling(6).mean()
df['avg_loss(6)']=df['loss'].rolling(6).mean()
df['avg_loss(6)'] = df['avg_loss(6)'].replace(0, 0.01)
df['rs_6']=df['avg_gain(6)']/df['avg_loss(6)']
df['rs_6'] = df['rs_6'].clip(0, 100) # 限制在0-100
df['rsi_6']=100-(100/(1+df['rs_6']))
#成交 MA5
df['money_ma5']=df['ma5']*df['vol_ma5']
#阳 1 阴 0
df['status']=np.where((df['close']-df['open'])>0,1,0)
#价pp
df['pp']=df['close']-df['close'].shift(1)
#量kk
df['kk']=df['volume']/df['vol_ma5']
# ==================== 【新增】MACD 标准公式 ====================
# 短期EMA(12)、长期EMA(26)、DIF、DEA(9)、MACD柱
df['ema12'] = df['close'].ewm(span=12, adjust=False).mean()
df['ema26'] = df['close'].ewm(span=26, adjust=False).mean()
df['dif'] = df['ema12'] - df['ema26'] # 快线
df['dea'] = df['dif'].ewm(span=9, adjust=False).mean() # 慢线
df['macd'] = 2 * (df['dif'] - df['dea']) # MACD柱
# ==================== 【新增】KDJ 标准公式 ====================
# ==================== 【修复 1】KDJ 分母为 0 问题 ====================
low_list = df['low'].rolling(9, min_periods=9).min()
high_list = df['high'].rolling(9, min_periods=9).max()
# 安全除法：分母为0时直接=0，永不爆炸
df['rsv'] = np.where(
high_list == low_list,
50,
(df['close'] - low_list) / (high_list - low_list) * 100
)
df['k'] = df['rsv'].ewm(com=2, adjust=False).mean()
df['d'] = df['k'].ewm(com=2, adjust=False).mean()
df['j'] = 3 * df['k'] - 2 * df['d']
# ==================== 【修复 3】清理所有异常值 ====================
df = df.replace([np.inf, -np.inf], np.nan)
df = df.fillna(0) # J=3K-2D
drop_cols = [
'hl', 'hc', 'lc', 'delta', 'gain', 'loss',
'avg_gain(14)', 'avg_loss(14)', 'avg_gain(6)', 'avg_loss(6)'
]
df = df.drop(columns=drop_cols, errors='ignore')
return df
#1 10收盘价接近10日最高点 10
# df.iloc[-1]['close']>(8*(df['high'].iloc[-10:].max()-df['high'].iloc[-10:].min())/10+df['high'].iloc[-10:].min())
# #2 收盘价大于MA5 10
# df.iloc[-1]['close']>df.iloc[-1]['ma5']
# #3 今日MA5>昨日MA5 10
# df.iloc[-1]['ma5']>df.iloc[-2]['ma5']
# #4 今日ATR在最近20个交易日内ATR排序 10
# atrr=(df['atr'].iloc[-20:]<df['atr'].iloc[-1]).sum()/20
# atrr>0.2 and atrr<0.8
# #5 平均成交额大于2亿 10
# df.iloc[-1]['money_ma5']>200000000
# #6 RSI 6小于70 10
# df.iloc[-1]['rsi(6)']<70
# df.iloc[-1]['rsi(6)']>30
# #7 收盘价大于MA20 10
# df.iloc[-1]['close']>df.iloc[-1]['ma20']
# #8 今日MA20>昨日MA20 10
# df.iloc[-1]['ma20']>df.iloc[-2]['ma20']
# #9 20日新高点 10
# df.iloc[-1]['high'] > df['high'].iloc[-20:-1].max()
# #10 量价指标 -100 -10 5 15
# p=df.iloc[-1]['close']-df.iloc[-2]['close']
# v=df.iloc[-1]['volume']/df['volume'].iloc[-20:].mean()
#p>0 2>v>1.5 +15
#p>0 v>1 +5
#p>0 v<1 -10 缩量
#p<0 v>2 -100 恐慌抛售
#p<=0 v<1 0 横盘
def score_def(df):
score=0
score_list=[]
atrr=(df['atr'].iloc[-20:]<df['atr'].iloc[-1]).sum()/20
p=df.iloc[-1]['close']-df.iloc[-2]['close']
v=df.iloc[-1]['volume']/df['volume'].iloc[-20:].mean()
if df.iloc[-1]['close']>(8*(df['high'].iloc[-10:].max()-df['high'].iloc[-10:].min())/10+df['high'].iloc[-10:].min()):
score=score+10
score_list.append("收盘价接近10日最高点 +10")
if df.iloc[-1]['close']>df.iloc[-1]['ma5']:
score=score+10
score_list.append("收盘价大于MA5 +10")
if df.iloc[-1]['ma5']>df.iloc[-2]['ma5']:
score=score+10
score_list.append("今日MA5>昨日MA5 +10")
if atrr>0.2 and atrr<0.8:
score=score+10
score_list.append("ATR在20日区间位置位于中部 +10")
if df.iloc[-1]['money_ma5']>200000000:
score=score+10
score_list.append("平均成交额大于2亿 +10")
if df.iloc[-1]['rsi(6)']<70 and df.iloc[-1]['rsi(6)']>30:
score=score+10
score_list.append("RSI6小于70大于30 +10")
if df.iloc[-1]['close']>df.iloc[-1]['ma20']:
score=score+10
score_list.append("收盘价大于MA20 +10")
if df.iloc[-1]['ma20']>df.iloc[-2]['ma20']:
score=score+10
score_list.append("今日MA20>昨日MA20 +10")
if df.iloc[-1]['high'] > df['high'].iloc[-20:-1].max():
score=score+10
score_list.append("20日新高点 +10")
if p > 0: # 上涨
if v > 1.5:
score += 15
score_list.append("健康上涨 +15")
elif v >= 1: # 1 <= v <= 1.5
score += 5
score_list.append("温和放量 +5")
else: # v < 1
score -= 10
score_list.append("缩量背离 -10")
else: # p <= 0 下跌或平盘
if v >= 2:
score -= 20
score_list.append("恐慌下跌 -20")
elif v >= 1: # 1 <= v < 2
score -= 10
score_list.append("放量下跌 -10")
else: # v < 1
score += 0
score_list.append("缩量整理 0")
return [score,score_list,df.iloc[-1].to_dict(),df.iloc[-5:].to_dict('records')]
# gp=GP()
# co=gp.gp_code('601515')
# df=gp.gupiao(co[0],co[1],days=60)
# gp.logout()
# df=gp.quant(df)
# for i in range(10):
# print(score_def(df))
# df=df.drop(df.index[-1])插入数据库：多条插入
import pymysql
from pymysql.err import OperationalError, ProgrammingError
from gp import GP
import pandas as pd
import time
def df_to_datalist(df):
insert_list = []
for _, row in df.iterrows():
data = (
row['code'][3:], row['code'], row['date'], row['open'], row['high'], row['low'], row['close'], row['preclose'], row['volume'], row['pctChg'],
row['turn'], row['isST'], row['adjustflag'], row['ma5'], row['ma10'], row['ma20'], row['ma60'], row['vol_ma5'], row['vol_ma10'], row['vol_ma20'],
row['vol_ma60'], row['tr'], row['atr'], row['rs_6'], row['rsi_6'], row['rs_14'], row['rsi_14'], row['money_ma5'], row['status'], row['pp'],
row['kk'], row['ema12'], row['ema26'], row['dif'], row['dea'], row['macd'], row['rsv'], row['k'], row['d'], row['j']
)
insert_list.append(data)
return insert_list
# 数据库配置
config = {
"host": "localhost",
"port": 3306,
"user": "root",
"password": "123456",
"database": "db",
"charset": "utf8mb4"
}
conn = pymysql.connect(**config)
cursor = conn.cursor()
# ===================== ✅ 核心：判断是否已存在 =====================
def is_exists(code, date):
"""根据 code + date 判断数据是否已存在"""
sql = "SELECT 1 FROM cmf_quant WHERE code = %s AND date = %s LIMIT 1"
cursor.execute(sql, (code, date))
return cursor.fetchone() is not None
# ===================== ✅ 改造：只插入不存在的数据 =====================
def insert_data(data_list):
sql = """
INSERT INTO `cmf_quant`
( `mcode`, `code`, `date`, `open`, `high`, `low`, `close`, `preclose`, `volume`, `pctChg`
, `turn`, `isST`, `adjustflag`, `ma5`, `ma10`, `ma20`, `ma60`, `vol_ma5`, `vol_ma10`, `vol_ma20`
, `vol_ma60`, `tr`, `atr`, `rs_6`, `rsi_6`, `rs_14`, `rsi_14`, `money_ma5`, `status`, `pp`
, `kk`, `ema12`, `ema26`, `dif`, `dea`, `macd`, `rsv`, `k`, `d`, `j`
)
VALUES
(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s
,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s
,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s
,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s
) """
new_data = []
skip_count = 0
# 逐条判断：存在就跳过，不存在就加入插入列表
for item in data_list:
code = item[1] # 对应 code 字段位置
date = item[2] # 对应 date 字段位置
if is_exists(code, date):
skip_count += 1
continue
new_data.append(item)
# 有新数据才插入
if new_data:
cursor.executemany(sql, new_data)
conn.commit()
return len(new_data), skip_count
# GP 初始化
gp = GP()
def do_code(exchange, code, days=500):
start = time.time()
df = gp.get_history(exchange, code, days=days)
df = gp.quant(df)
df = df.fillna(0)
df = df.astype(object).where(pd.notnull(df), None)
data_list = df_to_datalist(df)
insert_count, skip_count = insert_data(data_list)
print(f"股票 {code} → 插入:{insert_count}条 | 跳过:{skip_count}条 | 耗时:{time.time()-start:.2f}s")
# 批量执行
df = pd.read_csv('all.csv', dtype={'code': str})
for index, row in df.iterrows():
do_code(row['exchange'], row['code'], days=500)废弃：单条测试插入
import pymysql
from pymysql.err import OperationalError, ProgrammingError
from gp import GP
import pandas as pd
import time
def df_to_datalist(df):
"""把量化DataFrame转为数据库插入需要的列表格式"""
insert_list = []
for _, row in df.iterrows():
data = (
row['code'][3:], row['code'], row['date'], row['open'], row['high'], row['low'], row['close'], row['preclose'], row['volume'], row['pctChg'],
row['turn'], row['isST'], row['adjustflag'], row['ma5'], row['ma10'], row['ma20'], row['ma60'], row['vol_ma5'], row['vol_ma10'], row['vol_ma20'],
row['vol_ma60'], row['tr'], row['atr'], row['rs_6'], row['rsi_6'], row['rs_14'], row['rsi_14'], row['money_ma5'], row['status'], row['pp'],
row['kk'], row['ema12'], row['ema26'], row['dif'], row['dea'], row['macd'], row['rsv'], row['k'], row['d'], row['j']
)
insert_list.append(data)
return insert_list
# 1. 数据库连接配置（改成你自己的）
config = {
"host": "localhost", # 主机地址
"port": 3306, # 端口
"user": "root", # 用户名
"password": "123456", # 密码
"database": "db", # 数据库名
"charset": "utf8mb4" # 编码
}
conn = pymysql.connect(**config)
cursor = conn.cursor()
def insert_data(data_list):
sql = """
INSERT INTO `cmf_quant`
( `mcode`, `code`, `date`, `open`, `high`, `low`, `close`, `preclose`, `volume`, `pctChg`
, `turn`, `isST`, `adjustflag`, `ma5`, `ma10`, `ma20`, `ma60`, `vol_ma5`, `vol_ma10`, `vol_ma20`
, `vol_ma60`, `tr`, `atr`, `rs_6`, `rsi_6`, `rs_14`, `rsi_14`, `money_ma5`, `status`, `pp`
, `kk`, `ema12`, `ema26`, `dif`, `dea`, `macd`, `rsv`, `k`, `d`, `j`
)
VALUES
(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s
,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s
,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s
,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s
) """
# cursor.execute(sql, data)
cursor.executemany(sql, data_list)
conn.commit()
return cursor.rowcount
gp=GP()
c=time.time()
df=gp.get_history("sh",600000,days=180)
print("GET",time.time()-c)
c=time.time()
df=gp.quant(df)
print("QUANT",time.time()-c)
df = df.fillna(0) # 把所有空值 NaN 替换为 0
df = df.astype(object).where(pd.notnull(df), None) # 兼容数据库
print(len(df))
data_list=df_to_datalist(df)
insert_data(data_list)评分调整规则
rules=[
{
"ru":1,
"check":lambda df:df.iloc[-1]['close']>df.iloc[-1]['ma5'],
"score":6,
"name":"【短线】收盘价站上MA5"
},
{
"ru":2,
"check":lambda df:df.iloc[-1]['ma5']>df.iloc[-2]['ma5'],
"score":6,
"name":"【趋势】MA5均线向上拐头"
},
{
"ru":3,
"check":lambda df:df.iloc[-1]['money_ma5']>200000000,
"score":7,
"name":"【资金】5日均成交额超2亿"
},
{
"ru":4,
"check":lambda df:df.iloc[-1]['close']>(8*(df['high'].iloc[-10:].max()-df['high'].iloc[-10:].min())/10+df['high'].iloc[-10:].min()),
"score":7,
"name":"【位置】收盘价靠近10日高位"
},
{
"ru":5,
"check":lambda df:30<df.iloc[-1]['rsi_6']<70,
"score":5,
"name":"【指标】RSI(6)处于健康区间"
},
{
"ru":6,
"check":lambda df:df.iloc[-1]['close']>df.iloc[-1]['ma20'],
"score":7,
"name":"【趋势】收盘价站上MA20"
},
{
"ru":7,
"check":lambda df:df.iloc[-1]['ma20']>df.iloc[-2]['ma20'],
"score":7,
"name":"【趋势】MA20均线向上"
},
{
"ru":8,
"check":lambda df:df.iloc[-1]['high'] > df['high'].iloc[-20:-1].max(),
"score":12,
"name":"【强势】创20日新高"
},
{
"ru":9,
"check":lambda df:0.2<((df['atr'].iloc[-20:]<df['atr'].iloc[-1]).sum()/20)<0.8,
"score":5,
"name":"【波动】ATR波动率处于中段"
},
{
"ru":10,
"check":lambda df:df['pp'].iloc[-1]>0 and 1.5<=df['kk'].iloc[-1]<2 ,
"score":18,
"name":"【健康】量价配合稳步上涨"
},
{
"ru":11,
"check":lambda df:df['pp'].iloc[-1]>0 and 1<=df['kk'].iloc[-1]<1.5 ,
"score":10,
"name":"【温和】量能平稳温和上涨"
},
{
"ru":12,
"check":lambda df:df['pp'].iloc[-1]>0 and df['kk'].iloc[-1]>=2 ,
"score":-12,
"name":"【警示】倍量上涨（抛压大）"
},
{
"ru":13,
"check":lambda df:df['pp'].iloc[-1]<=0 and df['kk'].iloc[-1]>=2 ,
"score":-25,
"name":"【恐慌】放量暴跌（踩踏出逃）"
},
{
"ru":14,
"check":lambda df:df['pp'].iloc[-1]<=0 and 1<=df['kk'].iloc[-1]<2 ,
"score":-15,
"name":"【走弱】放量下跌（资金出逃）"
},
{
"ru":15,
"check":lambda df:df['pp'].iloc[-1]<=0 and df['kk'].iloc[-1]<1 ,
"score":0,
"name":"【整理】缩量盘整（方向不明）"
},
{
"ru":16,
"check":lambda df:df['high'].iloc[-1]>df['high'].iloc[-90:-1].max() and df['close'].iloc[-1]<df['high'].iloc[-90:-1].max() ,
"score":-18,
"name":"【诱多】90日假突破（主力出货）"
},
{
"ru":17,
"check":lambda df:df['high'].iloc[-1]>df['high'].iloc[-90:-1].max() and df['close'].iloc[-1]>df['high'].iloc[-90:-1].max() ,
"score":20,
"name":"【突破】有效突破90日高位"
},
{
"ru":18,
"check":lambda df:df['high'].iloc[-1]>=0.97*df['high'].iloc[-90:-1].max() and df['close'].iloc[-1]>=0.97*df['high'].iloc[-90:-1].max() ,
"score":10,
"name":"【压力】临近90日压力位"
},
{
"ru":19,
"check": lambda df: (
(df["isST"].iloc[-1] == 1 and df["pctChg"].iloc[-1] > 4.9) or
(df["code"].iloc[-1].startswith(("sh.600","sz.000","sz.001","sz.002")) and df["pctChg"].iloc[-1] > 9.9) or
(df["code"].iloc[-1].startswith(("sh.688","sz.300","sz.301")) and df["pctChg"].iloc[-1] > 19.8) or
(df["code"].iloc[-1].startswith(("bj.83","bj.87","bj.88","bj.92")) and df["pctChg"].iloc[-1] > 29.7)
),
"score": 20,
"name":"【涨停】当日强势涨停"
},
{
"ru":20,
"check": lambda df: (
((df["isST"].iloc[-1] == 1 and df["pctChg"].iloc[-1] > 4.9) or
(df["code"].iloc[-1].startswith(("sh.600","sz.000","sz.001","sz.002")) and df["pctChg"].iloc[-1] > 9.9) or
(df["code"].iloc[-1].startswith(("sh.688","sz.300","sz.301")) and df["pctChg"].iloc[-1] > 19.8) or
(df["code"].iloc[-1].startswith(("bj.83","bj.87","bj.88","bj.92")) and df["pctChg"].iloc[-1] > 29.7))
and (df.iloc[-1]["high"] - df.iloc[-1]["low"])/df.iloc[-1]["close"] <= 0.03
),
"score": 8,
"name":"【超强】缩量窄幅涨停（溢价高）"
},
{
"ru":21,
"check": lambda df: (
((df["isST"].iloc[-1] == 1 and df["pctChg"].iloc[-1] > 4.9) or
(df["code"].iloc[-1].startswith(("sh.600","sz.000","sz.001","sz.002")) and df["pctChg"].iloc[-1] > 9.9) or
(df["code"].iloc[-1].startswith(("sh.688","sz.300","sz.301")) and df["pctChg"].iloc[-1] > 19.8) or
(df["code"].iloc[-1].startswith(("bj.83","bj.87","bj.88","bj.92")) and df["pctChg"].iloc[-1] > 29.7))
and (df.iloc[-1]["high"] - df.iloc[-1]["low"])/df.iloc[-1]["close"] > 0.07
),
"score": -35,
"name":"【分歧】烂板涨停（封板无力）"
},
{
"ru":22,
"check": lambda df: (
(df["isST"].iloc[-1] == 1 and df["pctChg"].iloc[-1] < -4.9) or
(df["code"].iloc[-1].startswith(("sh.600","sz.000","sz.001","sz.002")) and df["pctChg"].iloc[-1] < -9.9) or
(df["code"].iloc[-1].startswith(("sh.688","sz.300","sz.301")) and df["pctChg"].iloc[-1] < -19.8) or
(df["code"].iloc[-1].startswith(("bj.83","bj.87","bj.88","bj.92")) and df["pctChg"].iloc[-1] < -29.7)
),
"score": -35,
"name":"【致命】当日跌停（空头主导）"
},
{
"ru":23,
"check":lambda df:df['isST'].iloc[-1]==1 ,
"score":-30,
"name":"【警示】ST股票（风险高）"
},
{
"ru":24,
"check":lambda df:df['turn'].iloc[-1]>=25 ,
"score":-60,
"name":"【警示】超高换手率（风险及高）"
},
{
"ru":25,
"check":lambda df:15<=df['turn'].iloc[-1]<25 ,
"score":-20,
"name":"【警示】高换手率（风险高）"
},
{
"ru":26,
"check":lambda df:5<=df['turn'].iloc[-1]<15 ,
"score":10,
"name":"【健康】活跃换手率"
}
]评分使用 ：mysql导入pandas
df = pd.read_sql(sql, conn, params=(mcode,))
import pandas as pd
import pymysql
from rules import rules
# 数据库配置
config = {
"host": "localhost",
"port": 3306,
"user": "root",
"password": "123456",
"database": "db",
"charset": "utf8mb4"
}
# 1. 创建连接
conn = pymysql.connect(**config)
def get_info(mcode):
sql = "SELECT * FROM cmf_quant WHERE mcode = %s ORDER BY date ASC"
df = pd.read_sql(sql, conn, params=(mcode,))
return df
def get_score(mcode):
df=get_info(mcode)
if len(df)>90:
score=0
date=df.iloc[-1]['date']
score_text=[]
code=df.iloc[-1]['code']
for ru in rules:
if ru['check'](df):
score=score+ru['score']
score_text.append(ru['name'])
return [
code,
date,
score,
score_text,
df.iloc[-1].to_dict(),
df.iloc[-15:].to_dict('records')
]
return False
print(get_score(600000)[0:4])