brunoandClaude Sonnet 4.5
c553009853
新增功能： - 5个专业数据抓取器（公司信息、板块、财务、资金流、龙虎榜） - 一键执行脚本 run_all_fetchers.sh（完整/测试/自定义三种模式） - 扩展数据库Schema（11张新表） - 数据源迁移至免费API（东方财富、新浪、腾讯） 数据能力： - 公司基础信息：5,295只A股 - 板块数据：~400个板块 + 20万成分股关系 - 财务报表：利润表+资产负债表（最近4季） - 资金流：主力资金、大单、中单、小单（700天） - 龙虎榜：上榜记录+营业部明细（2年） - 总数据量：~395万条记录，3-5GB存储 技术亮点： - 断点续传支持 - UPSERT策略避免重复 - API限流控制 - 实时进度显示 - 错误容错机制 - 完整日志记录 文档更新： - README添加"全量数据抓取"章节 - 60+页技术方案文档 - 完整的使用指南和维护文档 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
347 lines
12 KiB
Python
347 lines
12 KiB
Python
"""
|
|
公司基础信息抓取器
|
|
从东方财富获取公司基础信息，包括：
|
|
- 公司全称、行业分类
|
|
- 主营业务、经营范围
|
|
- 上市信息、注册信息等
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
import requests
|
|
import psycopg2
|
|
import time
|
|
import json
|
|
from datetime import datetime
|
|
from typing import Dict, List, Optional
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
from app.config import get_settings
|
|
|
|
settings = get_settings()
|
|
|
|
|
|
class CompanyInfoFetcher:
|
|
"""公司信息抓取器"""
|
|
|
|
def __init__(self):
|
|
self.conn = self._connect_db()
|
|
self.session = requests.Session()
|
|
self.session.headers.update({
|
|
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36',
|
|
'Referer': 'http://emweb.securities.eastmoney.com/'
|
|
})
|
|
|
|
def _connect_db(self):
|
|
"""连接数据库"""
|
|
db_url = settings.database_url
|
|
db_url = db_url.replace('postgresql://', '')
|
|
user_pass, host_db = db_url.split('@')
|
|
user, password = user_pass.split(':')
|
|
host_port, db = host_db.split('/')
|
|
host, port = host_port.split(':')
|
|
|
|
return psycopg2.connect(
|
|
host=host,
|
|
port=port,
|
|
database=db,
|
|
user=user,
|
|
password=password
|
|
)
|
|
|
|
def _get_stock_code(self, symbol: str) -> str:
|
|
"""
|
|
转换股票代码为东方财富格式
|
|
600000 -> SH600000
|
|
000001 -> SZ000001
|
|
"""
|
|
if symbol.startswith('6') or symbol.startswith('688'):
|
|
return f'SH{symbol}'
|
|
else:
|
|
return f'SZ{symbol}'
|
|
|
|
def fetch_company_info(self, symbol: str) -> Optional[Dict]:
|
|
"""
|
|
从东方财富获取公司基础信息
|
|
|
|
API: http://emweb.securities.eastmoney.com/PC_HSF10/CompanySurvey/PageAjax
|
|
"""
|
|
stock_code = self._get_stock_code(symbol)
|
|
|
|
url = "http://emweb.securities.eastmoney.com/PC_HSF10/CompanySurvey/PageAjax"
|
|
params = {
|
|
'code': stock_code
|
|
}
|
|
|
|
try:
|
|
response = self.session.get(url, params=params, timeout=10)
|
|
if response.status_code != 200:
|
|
print(f"❌ {symbol}: HTTP {response.status_code}")
|
|
return None
|
|
|
|
data = response.json()
|
|
|
|
# 解析公司基础信息
|
|
company_info = {}
|
|
|
|
# 基础信息部分 - jbzl是列表，取第一个元素
|
|
if 'jbzl' in data and isinstance(data['jbzl'], list) and len(data['jbzl']) > 0:
|
|
jbzl = data['jbzl'][0]
|
|
company_info['full_name'] = jbzl.get('ORG_NAME', '')
|
|
company_info['english_name'] = jbzl.get('ORG_NAME_EN', '')
|
|
company_info['legal_representative'] = jbzl.get('LEGAL_PERSON', '')
|
|
company_info['register_address'] = jbzl.get('FOUND_ADDR', '')
|
|
company_info['office_address'] = jbzl.get('OFFICE_ADDR', '')
|
|
company_info['zipcode'] = jbzl.get('ZIPCODE', '')
|
|
company_info['website'] = jbzl.get('ORG_WEB', '')
|
|
company_info['email'] = jbzl.get('ORG_EMAIL', '')
|
|
|
|
# 上市信息
|
|
list_date_str = jbzl.get('LISTING_DATE', '')
|
|
if list_date_str:
|
|
try:
|
|
company_info['list_date'] = datetime.strptime(list_date_str, '%Y-%m-%d').date()
|
|
except:
|
|
company_info['list_date'] = None
|
|
|
|
# 交易所
|
|
if symbol.startswith('6') or symbol.startswith('688'):
|
|
company_info['exchange'] = 'SH'
|
|
else:
|
|
company_info['exchange'] = 'SZ'
|
|
|
|
# 上市板块
|
|
if symbol.startswith('688'):
|
|
company_info['list_board'] = '科创板'
|
|
elif symbol.startswith('300'):
|
|
company_info['list_board'] = '创业板'
|
|
elif symbol.startswith('8') or symbol.startswith('4'):
|
|
company_info['list_board'] = '北交所'
|
|
else:
|
|
company_info['list_board'] = '主板'
|
|
|
|
# 股本信息（单位：股）
|
|
total_capital = jbzl.get('TOTAL_SHARES', '') # 总股本
|
|
if total_capital:
|
|
try:
|
|
company_info['total_capital'] = float(total_capital)
|
|
except:
|
|
company_info['total_capital'] = None
|
|
|
|
# 行业分类 - 直接从jbzl获取
|
|
company_info['industry_csrc'] = jbzl.get('INDUSTRYCSRC1', '')
|
|
company_info['industry_sw_l1'] = jbzl.get('EM2016', '') # 东财行业分类
|
|
company_info['industry_sw_l2'] = '' # API未提供，留空
|
|
company_info['industry_sw_l3'] = '' # API未提供，留空
|
|
|
|
# 主营业务
|
|
company_info['main_business'] = jbzl.get('MAIN_BUSINESS', '')
|
|
company_info['business_scope'] = jbzl.get('BUSINESS_SCOPE', '')
|
|
else:
|
|
# 如果API返回为空，使用默认值
|
|
company_info['full_name'] = ''
|
|
company_info['english_name'] = ''
|
|
company_info['legal_representative'] = ''
|
|
company_info['register_address'] = ''
|
|
company_info['office_address'] = ''
|
|
company_info['zipcode'] = ''
|
|
company_info['website'] = ''
|
|
company_info['email'] = ''
|
|
company_info['list_date'] = None
|
|
company_info['total_capital'] = None
|
|
company_info['industry_csrc'] = ''
|
|
company_info['industry_sw_l1'] = ''
|
|
company_info['industry_sw_l2'] = ''
|
|
company_info['industry_sw_l3'] = ''
|
|
company_info['main_business'] = ''
|
|
company_info['business_scope'] = ''
|
|
|
|
company_info['symbol'] = symbol
|
|
company_info['is_active'] = True
|
|
|
|
return company_info
|
|
|
|
except Exception as e:
|
|
print(f"❌ {symbol}: {e}")
|
|
return None
|
|
|
|
def save_company_info(self, info: Dict) -> bool:
|
|
"""保存公司信息到数据库"""
|
|
try:
|
|
cursor = self.conn.cursor()
|
|
|
|
sql = """
|
|
INSERT INTO t_company_info (
|
|
symbol, full_name, english_name, legal_representative,
|
|
main_business, business_scope,
|
|
industry_csrc, industry_sw_l1, industry_sw_l2, industry_sw_l3,
|
|
list_date, list_board, exchange,
|
|
register_address, office_address, zipcode, website, email,
|
|
total_capital, is_active
|
|
) VALUES (
|
|
%(symbol)s, %(full_name)s, %(english_name)s, %(legal_representative)s,
|
|
%(main_business)s, %(business_scope)s,
|
|
%(industry_csrc)s, %(industry_sw_l1)s, %(industry_sw_l2)s, %(industry_sw_l3)s,
|
|
%(list_date)s, %(list_board)s, %(exchange)s,
|
|
%(register_address)s, %(office_address)s, %(zipcode)s, %(website)s, %(email)s,
|
|
%(total_capital)s, %(is_active)s
|
|
)
|
|
ON CONFLICT (symbol) DO UPDATE SET
|
|
full_name = EXCLUDED.full_name,
|
|
english_name = EXCLUDED.english_name,
|
|
legal_representative = EXCLUDED.legal_representative,
|
|
main_business = EXCLUDED.main_business,
|
|
business_scope = EXCLUDED.business_scope,
|
|
industry_csrc = EXCLUDED.industry_csrc,
|
|
industry_sw_l1 = EXCLUDED.industry_sw_l1,
|
|
industry_sw_l2 = EXCLUDED.industry_sw_l2,
|
|
industry_sw_l3 = EXCLUDED.industry_sw_l3,
|
|
list_date = EXCLUDED.list_date,
|
|
list_board = EXCLUDED.list_board,
|
|
exchange = EXCLUDED.exchange,
|
|
register_address = EXCLUDED.register_address,
|
|
office_address = EXCLUDED.office_address,
|
|
zipcode = EXCLUDED.zipcode,
|
|
website = EXCLUDED.website,
|
|
email = EXCLUDED.email,
|
|
total_capital = EXCLUDED.total_capital,
|
|
updated_at = NOW()
|
|
"""
|
|
|
|
cursor.execute(sql, info)
|
|
self.conn.commit()
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"❌ Save error for {info.get('symbol')}: {e}")
|
|
self.conn.rollback()
|
|
return False
|
|
|
|
def get_all_stocks(self) -> List[str]:
|
|
"""从daily_quotes表获取所有股票代码"""
|
|
cursor = self.conn.cursor()
|
|
cursor.execute("""
|
|
SELECT DISTINCT symbol
|
|
FROM daily_quotes
|
|
ORDER BY symbol
|
|
""")
|
|
return [row[0] for row in cursor.fetchall()]
|
|
|
|
def get_completed_stocks(self) -> List[str]:
|
|
"""获取已完成的股票"""
|
|
cursor = self.conn.cursor()
|
|
cursor.execute("""
|
|
SELECT symbol
|
|
FROM t_company_info
|
|
ORDER BY symbol
|
|
""")
|
|
return [row[0] for row in cursor.fetchall()]
|
|
|
|
def run(self, limit: Optional[int] = None):
|
|
"""
|
|
运行抓取任务
|
|
|
|
Args:
|
|
limit: 限制抓取数量（用于测试）
|
|
"""
|
|
print("="*80)
|
|
print(" 公司基础信息抓取")
|
|
print("="*80)
|
|
|
|
# 1. 获取所有股票
|
|
all_stocks = self.get_all_stocks()
|
|
print(f"\n📋 数据库中共有 {len(all_stocks)} 只股票")
|
|
|
|
# 2. 获取已完成的股票
|
|
completed = self.get_completed_stocks()
|
|
print(f"✅ 已完成 {len(completed)} 只")
|
|
|
|
# 3. 计算待抓取
|
|
pending = [s for s in all_stocks if s not in completed]
|
|
print(f"⏳ 待抓取 {len(pending)} 只")
|
|
|
|
if limit:
|
|
pending = pending[:limit]
|
|
print(f"🔧 测试模式: 只抓取前 {limit} 只\n")
|
|
|
|
if not pending:
|
|
print("\n✅ 所有股票信息已抓取完成！")
|
|
return
|
|
|
|
# 4. 开始抓取
|
|
print("\n开始抓取...")
|
|
print("-"*80)
|
|
|
|
success_count = 0
|
|
failed_count = 0
|
|
|
|
for i, symbol in enumerate(pending, 1):
|
|
try:
|
|
# 抓取信息
|
|
info = self.fetch_company_info(symbol)
|
|
|
|
if info:
|
|
# 保存到数据库
|
|
if self.save_company_info(info):
|
|
success_count += 1
|
|
print(f"[{i}/{len(pending)}] ✅ {symbol} {info.get('full_name', '')} - {info.get('industry_sw_l1', '')}")
|
|
else:
|
|
failed_count += 1
|
|
print(f"[{i}/{len(pending)}] ❌ {symbol} - 保存失败")
|
|
else:
|
|
failed_count += 1
|
|
print(f"[{i}/{len(pending)}] ❌ {symbol} - 抓取失败")
|
|
|
|
# 避免限流
|
|
time.sleep(0.5)
|
|
|
|
# 每100条显示进度
|
|
if i % 100 == 0:
|
|
progress = (i / len(pending)) * 100
|
|
print(f"\n📊 进度: {i}/{len(pending)} ({progress:.1f}%) | 成功: {success_count} | 失败: {failed_count}\n")
|
|
|
|
except KeyboardInterrupt:
|
|
print("\n\n⚠️ 用户中断，正在保存进度...")
|
|
break
|
|
except Exception as e:
|
|
failed_count += 1
|
|
print(f"[{i}/{len(pending)}] ❌ {symbol} - 异常: {e}")
|
|
|
|
# 5. 总结
|
|
print("\n" + "="*80)
|
|
print(" 抓取完成")
|
|
print("="*80)
|
|
print(f"✅ 成功: {success_count}")
|
|
print(f"❌ 失败: {failed_count}")
|
|
print(f"📊 总计: {success_count + failed_count}")
|
|
print("="*80)
|
|
|
|
def __del__(self):
|
|
"""清理资源"""
|
|
if hasattr(self, 'conn'):
|
|
self.conn.close()
|
|
|
|
|
|
def main():
|
|
"""主函数"""
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(description='抓取公司基础信息')
|
|
parser.add_argument('--limit', type=int, help='限制抓取数量（测试用）')
|
|
parser.add_argument('--test', action='store_true', help='测试模式（只抓取10只）')
|
|
|
|
args = parser.parse_args()
|
|
|
|
limit = None
|
|
if args.test:
|
|
limit = 10
|
|
elif args.limit:
|
|
limit = args.limit
|
|
|
|
fetcher = CompanyInfoFetcher()
|
|
fetcher.run(limit=limit)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|