Spaces:
Running
Running
import re | |
import sys | |
import os | |
import numpy as np | |
from collections import defaultdict | |
import pandas as pd | |
import time | |
# 如果使用 spaCy 进行 NLP 处理 | |
import spacy | |
# 如果使用某种情感分析工具,比如 Hugging Face 的模型 | |
from transformers import pipeline | |
# 还需要导入 pickle 模块(如果你在代码的其他部分使用了它来处理序列化/反序列化) | |
import pickle | |
from gensim.models import KeyedVectors | |
import akshare as ak | |
from gensim.models import Word2Vec | |
from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification | |
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) | |
from us_stock import * | |
# 强制使用 GPU | |
#spacy.require_gpu() | |
# 加载模型 | |
try: | |
nlp = spacy.load("en_core_web_md") | |
except OSError: | |
print("Downloading model 'en_core_web_md'...") | |
from spacy.cli import download | |
download("en_core_web_md") | |
nlp = spacy.load("en_core_web_md") | |
# 检查是否使用 GPU | |
print("Is NPL GPU used Preprocessing.py:", spacy.prefer_gpu()) | |
# 使用合适的模型和tokenizer | |
model_name = "ProsusAI/finbert" # 选择合适的预训练模型 | |
tokenizer = AutoTokenizer.from_pretrained(model_name) | |
sa_model = AutoModelForSequenceClassification.from_pretrained(model_name) | |
# 初始化情感分析器 | |
sentiment_analyzer = pipeline('sentiment-analysis', model=sa_model, tokenizer=tokenizer) | |
index_us_stock_index_INX = ak.index_us_stock_sina(symbol=".INX") | |
index_us_stock_index_DJI = ak.index_us_stock_sina(symbol=".DJI") | |
index_us_stock_index_IXIC = ak.index_us_stock_sina(symbol=".IXIC") | |
index_us_stock_index_NDX = ak.index_us_stock_sina(symbol=".NDX") | |
class LazyWord2Vec: | |
def __init__(self, model_path): | |
self.model_path = model_path | |
self._model = None | |
def model(self): | |
if self._model is None: | |
print(f"Loading Word2Vec model from path: {self.model_path}...") | |
self._model = KeyedVectors.load(self.model_path, mmap='r') | |
return self._model | |
def vector_size(self): | |
self.load_model() | |
return self.model.vector_size # 现在你可以正确访问 vector_size 属性 | |
def __getitem__(self, key): | |
return self.model[key] | |
def __contains__(self, key): | |
return key in self.model | |
# 加载预训练的 Google News Word2Vec 模型 | |
# 定义模型名称 | |
from huggingface_hub import hf_hub_download | |
import os | |
# 定义 Hugging Face 的 repository 信息 | |
repo_id = "fse/word2vec-google-news-300" # 替换为实际的仓库ID | |
filename = "word2vec-google-news-300.model" # 文件名 | |
# 确保本地保存目录存在 | |
#os.makedirs(local_model_path, exist_ok=True) | |
# 尝试从 Hugging Face 下载模型文件 | |
try: | |
print(f"Downloading {filename} from Hugging Face Hub...") | |
downloaded_path = hf_hub_download( | |
repo_id=repo_id, | |
filename=filename | |
) | |
downloaded_path_npy = hf_hub_download( | |
repo_id=repo_id, | |
filename="word2vec-google-news-300.model.vectors.npy" | |
) | |
print(f"Model downloaded to {downloaded_path}") | |
except Exception as e: | |
raise RuntimeError(f"Failed to download {filename} from Hugging Face Hub: {e}") | |
# 加载模型 | |
print(f"Loading Word2Vec model from {downloaded_path}...") | |
word2vec_model = LazyWord2Vec(downloaded_path) | |
def pos_tagging(text): | |
try: | |
doc = nlp(text) | |
tokens, pos_tags, tags = [], [], [] | |
for token in doc: | |
if token.is_punct or token.is_stop: | |
continue | |
tokens.append(token.text) | |
pos_tags.append(token.pos_) | |
tags.append(token.tag_) | |
except Exception as e: | |
print(f"Error in pos_tagging for text: {text[:50]}... Error: {str(e)}") | |
return "", "", "" | |
return tokens, pos_tags, tags | |
# 命名实体识别函数 | |
def named_entity_recognition(text): | |
try: | |
doc = nlp(text) | |
entities = [(ent.text, ent.label_) for ent in doc.ents] | |
except Exception as e: | |
print(f"Error in named_entity_recognition for text: {text[:50]}... Error: {str(e)}") | |
entities = [] | |
return entities or [("", "")] | |
# 处理命名实体识别结果 | |
def process_entities(entities): | |
entity_counts = defaultdict(int) | |
try: | |
for entity in entities: | |
etype = entity[1] # 取出实体类型 | |
entity_counts[etype] += 1 # 直接对实体类型进行计数 | |
# 将字典转化为有序的数组 | |
entity_types = sorted(entity_counts.keys()) | |
counts = np.array([entity_counts[etype] for etype in entity_types]) | |
except Exception as e: | |
print(f"Error in process_entities: {str(e)}") | |
counts = np.zeros(len(entities)) | |
entity_types = [] | |
return counts, entity_types | |
# 处理词性标注结果 | |
def process_pos_tags(pos_tags): | |
pos_counts = defaultdict(int) | |
try: | |
for pos in pos_tags: | |
pos_counts[pos[1]] += 1 # 使用POS标签(如NN, VB等) | |
# 将字典转化为有序的数组 | |
pos_types = sorted(pos_counts.keys()) | |
counts = np.array([pos_counts[pos] for pos in pos_types]) | |
except Exception as e: | |
print(f"Error in process_pos_tags: {str(e)}") | |
counts = np.zeros(len(pos_tags)) | |
pos_types = [] | |
return counts, pos_types | |
# 函数:获取文档向量 | |
def get_document_vector(words, model = word2vec_model): | |
try: | |
# 获取每个词的词向量,如果词不在模型中则跳过 | |
word_vectors = [model[word] for word in words if word in model] | |
# 对词向量进行平均,得到文档向量;如果没有词在模型中则返回零向量 | |
document_vector = np.mean(word_vectors, axis=0) if word_vectors else np.zeros(model.vector_size) | |
except Exception as e: | |
print(f"Error in get_document_vector for words: {words[:5]}... Error: {str(e)}") | |
document_vector = np.zeros(model.vector_size) | |
return document_vector | |
# 函数:获取情感得分 | |
def get_sentiment_score(text): | |
try: | |
# 直接将原始文本传递给 sentiment_analyzer,它会自动处理 tokenization | |
result = sentiment_analyzer(text, truncation=True, max_length=512)[0] | |
score = result['score'] if result['label'] == 'positive' else -result['score'] | |
except Exception as e: | |
print(f"Error in get_sentiment_score for text: {text[:50]}... Error: {str(e)}") | |
score = 0.0 | |
return score | |
def get_stock_info(stock_codes, history_days=30): | |
# 获取股票代码和新闻日期 | |
stock_codes = stock_codes | |
news_date = datetime.now().strftime('%Y%m%d') | |
# print(f"Getting stock info for {stock_codes} on {news_date}") | |
previous_stock_history = [] | |
following_stock_history = [] | |
previous_stock_inx_index_history = [] | |
previous_stock_dj_index_history = [] | |
previous_stock_ixic_index_history = [] | |
previous_stock_ndx_index_history = [] | |
following_stock_inx_index_history = [] | |
following_stock_dj_index_history = [] | |
following_stock_ixic_index_history = [] | |
following_stock_ndx_index_history = [] | |
def process_history(stock_history, target_date, history_days=history_days, following_days = 3): | |
# 如果数据为空,创建一个空的 DataFrame 并填充为 0 | |
if stock_history.empty: | |
empty_data_previous = pd.DataFrame({ | |
'开盘': [-1] * history_days, | |
'收盘': [-1] * history_days, | |
'最高': [-1] * history_days, | |
'最低': [-1] * history_days, | |
'成交量': [-1] * history_days, | |
'成交额': [-1] * history_days | |
}) | |
empty_data_following = pd.DataFrame({ | |
'开盘': [-1] * following_days, | |
'收盘': [-1] * following_days, | |
'最高': [-1] * following_days, | |
'最低': [-1] * following_days, | |
'成交量': [-1] * following_days, | |
'成交额': [-1] * following_days | |
}) | |
return empty_data_previous, empty_data_following | |
# 确保 'date' 列存在 | |
if 'date' not in stock_history.columns: | |
print(f"'date' column not found in stock history. Returning empty data.") | |
return pd.DataFrame([[-1] * 6] * history_days), pd.DataFrame([[-1] * 6] * following_days) | |
# 将日期转换为 datetime 格式,便于比较 | |
stock_history['date'] = pd.to_datetime(stock_history['date']) | |
target_date = pd.to_datetime(target_date) | |
# 找到目标日期的索引 | |
target_row = stock_history[stock_history['date'] == target_date] | |
if target_row.empty: | |
# 如果目标日期找不到,找到离目标日期最近的日期 | |
closest_date_index = (stock_history['date'] - target_date).abs().idxmin() | |
target_date = stock_history.loc[closest_date_index, 'date'] | |
target_row = stock_history[stock_history['date'] == target_date] | |
# 确保找到的目标日期有数据 | |
if target_row.empty: | |
return pd.DataFrame([[-1] * 6] * history_days), pd.DataFrame([[-1] * 6] * following_days) | |
target_index = target_row.index[0] | |
target_pos = stock_history.index.get_loc(target_index) | |
# 取出目标日期及其前history_days条记录 | |
previous_rows = stock_history.iloc[max(0, target_pos - history_days):target_pos + 1] | |
# 取出目标日期及其后3条记录 | |
following_rows = stock_history.iloc[target_pos + 1:target_pos + 4] | |
# 删除日期列 | |
previous_rows = previous_rows.drop(columns=['date']) | |
following_rows = following_rows.drop(columns=['date']) | |
# 如果 previous_rows 或 following_rows 的行数不足 history_days,则填充至 history_days 行 | |
if len(previous_rows) < history_days: | |
previous_rows = previous_rows.reindex(range(history_days), fill_value=-1) | |
if len(following_rows) < 3: | |
following_rows = following_rows.reindex(range(3), fill_value=-1) | |
# 只返回前history_days行,并只返回前6列(开盘、收盘、最高、最低、成交量、成交额) | |
previous_rows = previous_rows.iloc[:history_days, :6] | |
following_rows = following_rows.iloc[:following_days, :6] | |
return previous_rows, following_rows | |
if not stock_codes or stock_codes == ['']: | |
# 如果 stock_codes 为空,直接获取并返回大盘数据 | |
stock_index_ndx_history = get_stock_index_history("", news_date, 1) | |
stock_index_dj_history = get_stock_index_history("", news_date, 2) | |
stock_index_inx_history = get_stock_index_history("", news_date, 3) | |
stock_index_ixic_history = get_stock_index_history("", news_date, 4) | |
previous_ndx_rows, following_ndx_rows = process_history(stock_index_ndx_history, news_date, history_days) | |
previous_dj_rows, following_dj_rows = process_history(stock_index_dj_history, news_date, history_days) | |
previous_inx_rows, following_inx_rows = process_history(stock_index_inx_history, news_date, history_days) | |
previous_ixic_rows, following_ixic_rows = process_history(stock_index_ixic_history, news_date, history_days) | |
previous_stock_inx_index_history.append(previous_inx_rows.values.tolist()) | |
previous_stock_dj_index_history.append(previous_dj_rows.values.tolist()) | |
previous_stock_ixic_index_history.append(previous_ixic_rows.values.tolist()) | |
previous_stock_ndx_index_history.append(previous_ndx_rows.values.tolist()) | |
following_stock_inx_index_history.append(following_inx_rows.values.tolist()) | |
following_stock_dj_index_history.append(following_dj_rows.values.tolist()) | |
following_stock_ixic_index_history.append(following_ixic_rows.values.tolist()) | |
following_stock_ndx_index_history.append(following_ndx_rows.values.tolist()) | |
# 个股补零逻辑 | |
previous_stock_history.append([[-1] * 6] * history_days) | |
following_stock_history.append([[-1] * 6] * 3) | |
else: | |
for stock_code in stock_codes: | |
stock_code = stock_code.strip() | |
stock_history = get_stock_history(stock_code, news_date) | |
# 处理个股数据 | |
previous_rows, following_rows = process_history(stock_history, news_date) | |
previous_stock_history.append(previous_rows.values.tolist()) | |
following_stock_history.append(following_rows.values.tolist()) | |
# 处理大盘数据 | |
stock_index_ndx_history = get_stock_index_history("", news_date, 1) | |
stock_index_dj_history = get_stock_index_history("", news_date, 2) | |
stock_index_inx_history = get_stock_index_history("", news_date, 3) | |
stock_index_ixic_history = get_stock_index_history("", news_date, 4) | |
previous_ndx_rows, following_ndx_rows = process_history(stock_index_ndx_history, news_date, history_days) | |
previous_dj_rows, following_dj_rows = process_history(stock_index_dj_history, news_date, history_days) | |
previous_inx_rows, following_inx_rows = process_history(stock_index_inx_history, news_date, history_days) | |
previous_ixic_rows, following_ixic_rows = process_history(stock_index_ixic_history, news_date, history_days) | |
previous_stock_inx_index_history.append(previous_inx_rows.values.tolist()) | |
previous_stock_dj_index_history.append(previous_dj_rows.values.tolist()) | |
previous_stock_ixic_index_history.append(previous_ixic_rows.values.tolist()) | |
previous_stock_ndx_index_history.append(previous_ndx_rows.values.tolist()) | |
following_stock_inx_index_history.append(following_inx_rows.values.tolist()) | |
following_stock_dj_index_history.append(following_dj_rows.values.tolist()) | |
following_stock_ixic_index_history.append(following_ixic_rows.values.tolist()) | |
following_stock_ndx_index_history.append(following_ndx_rows.values.tolist()) | |
# 只返回第一支股票的数据 | |
break | |
return previous_stock_history, following_stock_history, \ | |
previous_stock_inx_index_history, previous_stock_dj_index_history, previous_stock_ixic_index_history, previous_stock_ndx_index_history, \ | |
following_stock_inx_index_history, following_stock_dj_index_history, following_stock_ixic_index_history, following_stock_ndx_index_history, | |
def lemmatized_entry(entry): | |
entry_start_time = time.time() | |
# Step 1 - 条目聚合 | |
lemmatized_text = preprocessing_entry(entry) | |
return lemmatized_text | |
# 1. 数据清理 | |
# 1.1 合并数据 | |
# 1.2 去除噪声 | |
# 1.3 大小写转换 | |
# 1.4 去除停用词 | |
# 1.5 词汇矫正与拼写检查 | |
# 1.6 词干提取与词形还原 | |
# 强制使用 GPU | |
# spacy.require_gpu() | |
# 加载模型 | |
nlp = spacy.load("en_core_web_md") | |
# 检查是否使用 GPU | |
print("Is NPL GPU used Lemmatized:", spacy.prefer_gpu()) | |
def preprocessing_entry(news_entry): | |
"""数据清理启动函数 | |
Args: | |
text (str): preprocessing后的文本 | |
Returns: | |
[str]]: 词干提取后的String列表 | |
""" | |
# 1.1 合并数据 | |
text = merge_text(news_entry) | |
# 1.2 去除噪声 | |
text = disposal_noise(text) | |
# 1.3 大小写转换 | |
text = text.lower() | |
# 1.4 去除停用词 | |
text = remove_stopwords(text) | |
# 1.5 拼写检查 | |
#text = correct_spelling(text) | |
#print(f"1.5 拼写检查后的文本:{text}") | |
# 1.6 词干提取与词形还原 | |
lemmatized_text_list = lemmatize_text(text) | |
#print(f"1.6 词干提取与词形还原后的文本:{lemmatized_text_list}") | |
return lemmatized_text_list | |
# 1.1 合并数据 | |
def merge_text(news_entry): | |
return news_entry | |
# 1.2 去除噪声 | |
def disposal_noise(text): | |
# 移除HTML标签 | |
text = re.sub(r'<.*?>', '', text) | |
# 移除URLs | |
text = re.sub(r'http\S+|www\S+|https\S+', '', text, flags=re.MULTILINE) | |
# 移除方括号内的内容 | |
# text = re.sub(r'\[.*?\]', '', text) | |
# 移除标点符号 | |
# text = re.sub(r'[^\w\s]', '', text) | |
# 移除多余的空格 | |
text = re.sub(r'\s+', ' ', text).strip() | |
# 或者选择性地过滤,例如移除表情符号 | |
# text = re.sub(r'[^\w\s.,!?]', '', text) | |
# 移除换行符和制表符 | |
text = re.sub(r'[\n\t\r]', ' ', text) | |
return text | |
# 1.4 去除停用词 | |
def remove_stopwords(text): | |
# 使用 spaCy 处理文本 | |
doc = nlp(text) | |
# 去除停用词,并且仅保留标识为“词”(Token.is_alpha)类型的标记 | |
filtered_sentence = [token.text for token in doc if not token.is_stop and (token.is_alpha or token.like_num)] | |
return ' '.join(filtered_sentence) | |
# 1.5 拼写检查 | |
# 该函数用于检查输入文本的拼写错误,并修正 | |
# def correct_spelling(text): | |
# corrected_text = [] | |
# doc = nlp(text) | |
# for token in doc: | |
# if token.is_alpha: # 仅检查字母构成的单词 | |
# corrected_word = spell.correction(token.text) | |
# if corrected_word is None: | |
# # 如果拼写检查没有建议,保留原始单词 | |
# corrected_word = token.text | |
# corrected_text.append(corrected_word) | |
# else: | |
# corrected_text.append(token.text) | |
# return " ".join(corrected_text) | |
# 1.6 词干提取与词形还原 | |
# 该函数用于对输入文本进行词形还原,返回一个包含词形还原后单词 | |
def lemmatize_text(text): | |
# 提取词干化后的词 | |
lemmatized_words = [] | |
doc = nlp(text) # 需要在这里处理输入文本 | |
for token in doc: | |
# 忽略标点符号和空格 | |
if not token.is_punct and not token.is_space and (token.is_alpha or token.like_num): | |
lemmatized_words.append(token.lemma_) | |
return lemmatized_words | |
# 2. 数据增强和特征提取 | |
# 2.1 词性标注(Part-of-Speech Tagging) | |
# 为每个词标注其词性(如名词、动词、形容词等),这有助于后续的句法分析和信息提取。 | |
# 工具:spaCy 或 NLTK | |
# 2.2 命名实体识别(NER) | |
# 识别文本中的命名实体,如人名、地名、组织机构等,提取出这些实体信息。 | |
# 工具:spaCy 或 Stanford NER | |
# 2.3 句法分析与依存分析 | |
# 分析句子结构,理解单词之间的关系(如主谓宾结构)。 | |
# 工具:spaCy 或 NLTK | |
# 2 特征提取 | |
# 强制使用 GPU | |
#spacy.require_gpu() | |
# 加载模型 | |
nlp = spacy.load("en_core_web_md") | |
# 检查是否使用 GPU | |
print("Is NPL GPU used Enchance_text.py:", spacy.prefer_gpu()) | |
# 2.3 句法分析与依存分析 | |
def dependency_parsing(text): | |
doc = nlp(text) | |
dependencies = [] | |
for token in doc: | |
# 过滤标点符号和停用词,或其他不需要的词性 | |
if token.is_punct or token.is_stop: | |
continue | |
# 可以进一步根据特定的依存关系类型过滤结果 | |
# 常见的依存关系类型: 'nsubj' (名词主语), 'dobj' (直接宾语), 等等 | |
# if token.dep_ not in {'nsubj', 'dobj', ...}: | |
# continue | |
dependencies.append((token.text, token.dep_, token.head.text)) | |
return dependencies | |
def processing_entry(entry): | |
# print(f"processing_entry: {entry}") | |
lemmatized_entry = preprocessing_entry(entry) | |
# print(f"lemmatized_entry: {lemmatized_entry}") | |
cleaned_text = disposal_noise(entry) | |
# print(f"disposal_noise: {cleaned_text}") | |
pos_tag = pos_tagging(cleaned_text) | |
# print(f"pos_tagging: {db_pos_tag}") | |
ner = named_entity_recognition(cleaned_text) | |
# print(f"named_entity_recognition: {db_ner}") | |
dependency_parsed = dependency_parsing(cleaned_text) | |
# print(f"dependency_parsing: {db_dependency_parsing}") | |
sentiment_score = get_sentiment_score(cleaned_text) | |
# print(f"sentiment_score: {sentiment_score}") | |
return (lemmatized_entry, pos_tag, ner, dependency_parsed, sentiment_score) | |