通过 Python 爬虫提高股票选股胜率
此贴为Python爬虫技术学习贴
在股票中，即便有了选股规则，从5000多只股票中筛选出符合规则的股票也是十分困难的，于是想通过爬虫来实现自动化的快速选股。全文用GP代替股票
实现方案
1、指定两套规则，第一套弱约束，第二套强约束
2、每天3点收盘后，使用弱约束条件，筛一次全量的GP池，筛出的GP作为第二天的GP筛选池使用
3、集合竞价结束，9点25分，使用强约束，从前一天收盘后选出的GP池中进一步筛选，选出当日股票（通过不断优化强弱约束条件，此时选出的GP拥有高胜率）
我的约束（自己研究的规则，因人而异）
弱约束：收盘后，整体趋势向上，筹码集中度高
强约束：当天竞价完成后，竞价强势，交易量大，竞价涨跌幅大，等规则
爬虫实现
用到的是python中的akshare，ak中有很多数据接口，调用出来很方便，作为ak的补充，如果有些功能ak不具备，就需要手写请求，从东财、新浪财经等这些网站上抓取数据，手写请求一般会用到selenium，模拟通过浏览器访问（因为数据在这些网站上都是动态的，无法直接通过request请求到数据，使用selenium更方便）
运行环境
Python 3.10
requests~=2.32.3
akshare~=1.15.83
pandas~=2.2.3
selenium~=4.28.1
bs4~=0.0.2
beautifulsoup4~=4.12.3
APScheduler~=3.11.0实现代码（部分）
1、通过selenium配置浏览器
# 设置 Chrome 配置
chrome_options = Options()
chrome_options.add_argument("--headless") # 无头模式，不打开浏览器窗口
chrome_options.add_argument("--no-sandbox")
chrome_options.add_argument("--disable-dev-shm-usage")
# 启动 Chrome 浏览器
service = Service(CHROME_DRIVER_URL) # 替换为 chromedriver 的路径
driver = webdriver.Chrome(service=service, options=chrome_options)2、获取筹码集中度
# 取90%筹码集中度
def get_ChouMa_Jizhongdu(driver, code):
url = 'https://quote.eastmoney.com/concept/' + exchange_detector(code) + code + '.html#chart-k-cyq'
# 打开网页
driver.get(url)
# 获取整个页面的文本
try:
# 获取分钟竞价情况
app_element = driver.find_element(By.ID, 'app')
maincharts_ele = app_element.find_element(By.CLASS_NAME, 'maincharts').text
match = re.search(r'90%成本:.*?集中度:\s*(\d+\.\d+)%', maincharts_ele, re.DOTALL)
if match:
concentration_value = match.group(1)
return float(concentration_value)
else:
return -1
except Exception as e:
print(f"发生错误: {e}")
return -13、获取某只GP所在行业
def get_hangye(code):
try:
# test = ak.stock_individual_info_em(symbol=code)
hy = ak.stock_individual_info_em(symbol=code).value[6]
return hy
except Exception as e:
print(f"发生错误: {e}")
return "未获取"4、获取均线，用于判断整体走势
def get_today_ma(stock_code="000001", ma_periods=[5, 10, 20, 60]):
"""
获取股票当日均线价格
:param stock_code: 股票代码（默认示例代码为平安银行 "000001"）
:param ma_periods: 均线周期列表（默认计算 5、10、20、60 日均线）
:return: 当日均线值的字典（若当日无数据，返回前一个交易日均线）
"""
# 获取股票历史行情数据（调整为最近 120 个交易日，确保足够计算长期均线）
df = ak.stock_zh_a_hist(symbol=stock_code, period="daily", adjust="qfq", timeout=(5, 10)).iloc[-120:]
# 检查数据是否为空
if df.empty:
raise ValueError("未获取到股票数据，请检查代码或网络连接")
# 计算均线
for period in ma_periods:
df[f'MA{period}'] = df['收盘'].rolling(window=period).mean()
# 获取最新数据（当日或最近交易日）
latest_data = df.iloc[-1]
# 提取均线值
ma_values = {
f'MA{period}': round(latest_data[f'MA{period}'], 2)
for period in ma_periods
}
return ma_values5、获取某只GP今日开盘、昨日收盘情况
def get_today_open_and_yesterday_close(stock_code="000001"):
"""
获取股票今日开盘价和昨日收盘价
:param stock_code: 股票代码（默认示例代码为平安银行 "000001"）
:return: 今日开盘价和昨日收盘价的字典
"""
# 获取历史行情数据（最近两个交易日）
hist_data = ak.stock_zh_a_hist(symbol=stock_code, period="daily", adjust="qfq", timeout=(5, 10)).iloc[-2:]
# 检查历史数据是否为空
if hist_data.empty:
raise ValueError(f"未找到股票代码为 {stock_code} 的历史数据")
# 获取昨日收盘价
yesterday_close = hist_data.iloc[0]["收盘"]
today_open = hist_data.iloc[1]["开盘"]
ratio = hist_data.iloc[1]["涨跌幅"]
today_close = hist_data.iloc[1]["收盘"]
return {
"今开": today_open,
"昨收": yesterday_close,
"涨跌幅": ratio,
"今收": today_close
}6、获取实时委差
def get_weicha(code, driver):
symbol = code
url = f'https://finance.sina.com.cn/realstock/company/{exchange_detector(symbol)}{symbol}/nc.shtml'
# 打开网页
driver.get(url)
# 获取整个页面的文本
try:
# 获取竞价情况
tabfive_element = driver.find_element(By.ID, 'fiveAmt')
value = tabfive_element.text
return int(value)
except Exception as e:
print(f"发生错误: {e}")
return -17、获取9：25分的竞价量能
def get_ln(code, driver, today_date):
symbol = code
url = 'https://vip.stock.finance.sina.com.cn/quotes_service/view/vMS_tradedetail.php?symbol=' + exchange_detector(
symbol) + symbol + '&date=' + today_date + '&page=' + str(get_page_num(symbol, today_date))
# 打开网页
driver.get(url)
# 获取整个页面的文本
try:
# 获取竞价情况
tbody_element = driver.find_element(By.CLASS_NAME, 'dataOuter').find_element(By.TAG_NAME, 'tbody')
last_tr = tbody_element.find_elements(By.TAG_NAME, 'tr')[-1].get_attribute('innerHTML')
print(last_tr)
# 使用字符串的split方法分割数据
parts = last_tr.split("<td>")
# 获取第5个<td>中的数据（索引为4，因为索引从0开始）
value = parts[4].split("</td>")[0]
return int(value)
except Exception as e:
print(f"发生错误: {e}")
return -18、获取实时GP涨跌幅排序
def get_page(url):
try:
response = requests.get(url)
return response.text
except requests.ConnectionError as e:
print('', e.args)
# 获取股票代码、名称、PE,最高价，最小价，市净率，市盈率
def get_stock_data(text):
# .* ?"f9": (?P < pe >.+?) 匹配市盈率 .*?"f23":(?P<pb>.+?) 匹配市净率 ，注意需要按照f1,f2...这样的顺序
com = re.compile(
'"f2":(?P<end>.+?),.*?"f6":(?P<volume>.+?),.*?"f9":(?P<pe>.+?),.*?"f12":(?P<number>.+?),.*?"f14":(?P<name>.+?)'
',.*?"f15":(?P<max>.+?),.*?"f16":(?P<min>.+?),.*?"f17":(?P<start>.+?),.*?"f23":(?P<pb>.+?),.*?"f24":(?P<a>.+?)',
re.S)
ret = com.finditer(text)
for i in ret:
yield {
'number': i.group('number'),
'name': i.group('name'),
'start': i.group('start'),
'max': i.group('max'),
'min': i.group('min'),
'end': i.group('end'),
'pe': i.group('pe'), # 解析获取市盈率
'pb': i.group('pb'), # 解析市净率
'a': i.group('a'),
'volume': i.group('volume')
}
# 开始页码，和结束解码
def get_code_pe(start=1, end=1):
# 将所有的股票代码放入列表中
b = []
for i in range(start, end + 1):
url = 'http://60.push2.eastmoney.com/api/qt/clist/get?cb=jQuery112408744624686429123_1578798932591&pn=' \
'%d&pz=20&po=1&np=1&ut=bd1d9ddb04089700cf9c27f6f7426281&fltt=2&invt=2&fid=f3&fs=m:0+t:6,m:0+t:13,m:' \
'0+t:80,m:1+t:2,m:1+t:23&fields=f1,f2,f3,f4,f5,f6,f7,f8,f9,f10,f12,f13,f14,f15,f16,f17,f18,f20,f21,' \
'f23,f24,f25,f22,f11,f62,f128,f136,f115,f152&_=1586266306109' % i
content = get_page(url=url)
data = get_stock_data(text=content)
for j in data:
# 定义获取股票代码，名字列表
a = []
number = j.get('number')
# 加入股票代码
a.append(number)
name = j.get('name')
# 加入股票名字
a.append(name)
start = j.get('start')
max_price = j.get('max')
min_price = j.get('min')
end = j.get('end')
volume = j.get('volume')
pe = j.get('pe')
# 加入市盈率
a.append(pe)
pb = j.get('pb')
# 加入市盈率
a.append(pb)
if start == '"-"':
start, max_price, min_price, end, volume, pe, pb = '0', '0', '0', '0', '0', '0', '0'
b.append(a)
print(len(b))
return b
# 返回：所有GP列表，列表按实时涨跌幅排序
def get_stock_codes(end):
lt = get_code_pe(1, end)
return lt