分类： 人工智能

上一篇写到如何用Python计算指标，但是仅仅用这些常见指标，来对股票走势进行判断，胜率并不高，原因很简单，这些指标所有人都在看，就容易被庄家操控，另外，所有指标都只是人类对股票价格数据的理解，是不够全面的，毕竟人类在面对海量数据时的能力是远低于机器的。

梯度提升决策树HistGradientBoosting算法，我同时测试了随机森林、xgboost、lightGBM等算法，它给我的印象是又快又好，具体算法可以自行去网上搜，这里就不赘述了。网上看了很多用机器学习对股价进行预测的文章，大部分都是基于对收盘价序列，进行下一天的收盘价的预测，这种方法基本属于痴人说梦，只要写一个下一天收盘价等于上一天收盘价的函数，就能战胜绝大多数模型。

一般情况下，我们对股价趋势进行预测，不是只看收盘价，也不是只看前一天的收盘价，而是会看很多天的开盘价、收盘价、最高价和最低价，同时也要看成交量，成交量可能还更重要。（当然，还有更多的东西要看，比如市场情绪、政策趋向、突发事件、板块走势等等，这些数据很多都是非结构化的文本数据，所以大模型才是正解，当然，我们可以认为，这些数据，已经部分的反应在了股票成交数据中）更进一步，如果要分析tick数据（分笔的成交数据），理论上讲可以获得更多信息，而且可以更实时而精准地找到买卖点，但是，投入有点大，我的笔记本存不下，也跑不动，更重要的是，我只是写着玩，并不是真正去做股票投资，所以用日成交数据足够了。

所以，数据用的是N天的收盘价、开盘价、最高价、最低价、换手率，又加上了5日、10日、20日和100日的均线，用来反映更长期的变化信息。这里N天，到底是用10天数据去预测，还是用30天数据去预测，是个重要的超参数，虽然理论上讲N越大，预测越准，但是同样会带来性能问题。然后把这些数据直接平铺，当做训练的数据集。停牌时的数据，自动用上一日的填充。这里要说明的是HistGradientBoosting是支持缺失值的，但是考虑到更多其他算法使用，还是做了填充。同样，理论上讲HistGradientBoosting是不需要对数据做归一化的，但是不太股票股价差异太大，不归一化的话，模型泛化能力会非常差，没有用sklearn的minmaxscaler，因为发现它归一化出来的数字会损失掉非常多。这里，直接用每个股票价格的平均值当分母，去除股价，来做归一化，每只股票都是用自己的归一化算法，保证出来的数值，各个股票之间不会差太多，换手率不用做调整，

第二个问题是，为什么是预测趋势，而不是预测股价。这里考虑的其实是，到底用分类，还是用回归。预测股价的问题是，股价的影响因素太多，不够稳定，经常一涨一跌，对模型的稳定性不利，趋势就稳定很多，所以这个算法里面，预测的是5日均线在5天后比现在高10%。当然， 到底用5日还是7日均线，到底比较的是10天还是15天，到底高10%还是20%，这些都可以调，都是超参数。不过，理论上讲，这些参数选得越值越大，预测的准确性会越低，所以这个算法最好还是做短线。

下面上代码：

def check_buy_sell(a, b):
    #print('a='+str(a)+' b='+str(b))
    if b > a * 1.10:
        return 1
    elif b < a * 0.90:
        return -1
    return 0
    
def prepareStockData(stock_id, days, end_date, n_days):

    #多取一些时间，保证均线计算正确
    df = get_stock_hist_data(stock_id, start_date='2018-01-01', end_date=end_date)

    df['SMA_5_5'] = df['SMA_5'].shift(periods=-5)
    df['buysell'] = df.apply(lambda x: check_buy_sell(x['SMA_5'], x['SMA_5_5']), axis=1)

    #再截取要用的数据
    df = df[len(df)-n_days-days-5:len(df)-5]

    #print(df[['SMA_5','SMA_5_5']])

    samplesX = pd.DataFrame()
    samplesY = pd.DataFrame()

    #跳过前面几天数据，从第days天开始取
    for i in range(days, len(df)):
   
        df_sample_X = df.iloc[i-days:i] #取前days天到前一天数据
        df_sample_Y = df.iloc[i:i+1]    #取当天数据
            
        if not samplesX.empty:
            samplesX = pd.concat([samplesX, df_sample_X])
            samplesY = pd.concat([samplesY, df_sample_Y])
        else:
            samplesX = df_sample_X.copy()
            samplesY = df_sample_Y.copy()

    size = len(samplesY)
  

    X = np.array(samplesX[['open','close','high','low','SMA_5','SMA_10','SMA_20','SMA_100']]).reshape(size, days*8)
    exchanges = np.array(samplesX[['exchange']]).reshape(size,days)
    
    Y = np.array(samplesY[['buysell']]).reshape(size,1) #.ravel()
    ScaleNum = np.mean(X.reshape(size*days*8, 1))


    X = X / ScaleNum

    X = np.concatenate((X, exchanges), axis=1)

    return X, Y, ScaleNum

然后开始进行训练，同时训练了一个随机森林的模型，用来比较时间和准确率。这里有两点，在调参时，训练数据要截断在某个时间，以防止模型中包含了测试集的信息。二是训练好的模型，连同归一化参数，都用joblib保存下来，方便预测时调用。

def prepareTrainData(stock_ids, days, end_date='2025-08-30'):

    ALL_X = np.array([])
    ALL_Y = np.array([])
    scalers = {}

    for index, row in stock_ids.iterrows():
        stock_id = row['stock_id']
        print('getting train data for stock_id = '+stock_id)
        
        X, Y, scaleNum = prepareStockData(stock_id, days, end_date=end_date, n_days=1000)
        scalers[stock_id] = scaleNum
        
        if len(ALL_X) > 0:
            ALL_X = np.vstack((ALL_X, X))
            ALL_Y = np.vstack((ALL_Y, Y))
        else:
            ALL_X = X
            ALL_Y = Y

    return ALL_X, ALL_Y, scalers

def trainAndSaveModel(days=60, end_date='2025-08-30'):
    
    stock_ids = read_stock_pool_from_file(POOL_FILE)
    #today = str(get_last_trading_day())

    ALL_X, ALL_Y, scalers = prepareTrainData(stock_ids, days=days, end_date=end_date)

    print(ALL_X.shape)
    print(ALL_Y.shape)

    n_samples, n_length = ALL_X.shape
    n_samples = round(n_samples * 0.75)

    X_train = ALL_X[:n_samples]
    X_test  = ALL_X[n_samples:]
    Y_train = ALL_Y[:n_samples]
    Y_test  = ALL_Y[n_samples:]

    print('start fit model random forest, time=' + str(datetime.datetime.now()))
    
    model_randomForest = RandomForestClassifier().fit(X_train, Y_train.ravel())
    print(model_randomForest.score(X_test, Y_test.ravel()))
    print('start fit model hist gradient boosting, time=' + str(datetime.datetime.now()))
    
    model_histGradientBoosting = HistGradientBoostingClassifier(loss='log_loss',max_iter=100).fit(X_train, Y_train.ravel())
    print(model_histGradientBoosting.score(X_test, Y_test.ravel()))
    print('end fit, time=' + str(datetime.datetime.now()))

    saveDir = 'models_20250924'
    os.mkdir(saveDir)
    
    joblib.dump(model_randomForest, saveDir+'/model_randomForest.joblib')
    joblib.dump(model_histGradientBoosting, saveDir+'/model_histGradientBoosting.joblib')
    joblib.dump(scalers, saveDir+'/scalers.joblib')

    return model_randomForest, model_histGradientBoosting, scalers

这个算法，用自己写的回测系统，有非常好而且稳定的收益率，这个后面再讲，包括怎么写一个自己的回测系统。

在医疗AI项目中遇到要LLM基于自己的知识库来输出结果，综合考虑，决定采用PostgreSQL实现自己的RAG库。

关于技术选型

主要是觉得AI能力落地应用、或者企业内部时，大部分企业应该做好的是系统的整合，而不是对AI本身的调优。更关注企业特有的数据，更重要的是数据的独特性，增强模型的容错性，而不是过于关注数据的清洗和数据的数量。举个例子，在医疗论文搜索中，对标题、摘要、正文和引用，可以用不同的方式进行查询和使用；或者对特定内容，比如公司介绍、产品介绍、治疗方案介绍，有不同的查询方式等等。

先说为什么要选这样的技术方案：

1，能用RAG的，就坚决不用微调。RAG的成本远低于微调，而且保留足够的灵活度，便于后续替换和升级。目前各大LLM模型的上下文窗口越来越长，效果越来越好，价格也越来越低，绝大部分情况下，已经没有必要去自己微调、部署；

2，项目不大，对postgreSQL又比较熟悉，研究了一下pg的全文搜索和pgvector两个插件，发现够用，而且本来在项目中已经有了一个pg数据库，就直接拿来用，不用额外成本。

关于全文搜索

再说说为什么在使用语义搜索的同时，还要考虑全文搜索的支持，我想主要是几个原因：一个是项目实现阶段，可以使用全文搜索对语义搜索的结果进行校验；一个是实际应用种，可以使用全文搜索对语义搜索出来的内容，调整权重，实现一些特殊的需求。

既然用了全文搜索，那么就有分词的问题，项目里使用的是jieba分词，直接用pg的插件支持就可以，还能够自定义字典，非常好用。可以直接在SQL语句中完成分词：

SELECT  to_tsvector('jiebacfg', %s);

要注意的一点是，项目中文本可能会有英文，也会有中文，遇到英文时，要用英文的分词，写成这样

SELECT  to_tsvector('english', %s);

举个表结构的字段，article_part_ts就是to_tsvector存储字段，同时建立一个索引：

CREATE TABLE public.fy_article_part_index (
	fy_article_part_id serial4 NOT NULL,
	fy_article_info_id int4 NOT NULL,
	article_part_text text NOT NULL,
	article_part_ts tsvector NOT NULL,
	article_part_em public.vector NOT NULL,
	CONSTRAINT fy_article_part_index_pk PRIMARY KEY (fy_article_part_id)
);

CREATE INDEX fy_article_part_ts_index ON public.fy_article_part_index USING rum (article_part_ts);

搜索时，同样需要对待搜索内容进行分词，然后对分词内容进行组合（与、或都可以，效果差别很大，在不同场景下适用），再对搜索结果相关性排序，取前N个结果就可以，比如：

SELECT fy_article_info_id, article_part_ts <=> to_tsquery('jiebacfg', %s) AS rank

FROM  fy_article_part_index
WHERE article_part_ts @@ to_tsquery('jiebacfg', %s)
ORDER BY rank
LIMIT 100

关于语义搜索

其实前面的创建表的SQL中，已经包含了语义搜索用的字段，也就是article_part_em字段，格式是public.vector。

这里先说embedding，他是用一个高维向量来表示一个对象的内涵意义，常见的word2vec，或者openai提供的’text-embedding-ada-002’等，都是embedding。项目里，我直接使用了阿里提供的embedding服务接口（不要自己去训练，完全没必要），不过要注意，一般大家用的embedding都是1536个维度的，但是阿里不知道为什么，把最新的模型改成了1024维度，为了保留项目的兼容性，我使用了它上一个版本，也就是1536维度的版本。记得对存储embedding的字段加上索引：

CREATE INDEX fy_article_part_em_index ON public.fy_article_part_index USING hnsw (article_part_em vector_l2_ops) WITH (m='16', ef_construction='64');

查询的时候，也先调用接口对待查询内容进行embedding，然后执行SQL：

SELECT fy_article_info_id, article_part_em <#> %s::vector AS rank
FROM  fy_article_part_index
ORDER BY rank
LIMIT 100

这里要注意，我们知道两个矢量之间的距离有好几种计算方法，常见的包括：欧式距离、曼哈顿距离、余弦距离、内积等等，在项目应用中，略微有些差别，且计算开销也不一样。我这里用的操作符<#>计算的是内积，表示两个向量的相似度。

关于文本切片

无论是从数据存储、搜索还是收API服务的接口限制上来说，我们都要对文本内容进行切片。

切片可以固定长度切，也可以根据语句来切，切片的长度太短会造成语义搜索效果变差，长度太长，又会造成性能问题，所以要结合项目进行调优。

我采取的策略是先进行分句，然后把长度限制内的语句，组合成一个切片。

对于英文内容，项目里选择的分句方法是Python的NLTK库：

from nltk.tokenize import sent_tokenize

sentences = sent_tokenize(article_en)

对于中文，没有找到现成的分句方法，就自己实现了一个，考虑了一些特殊情况

def split_chinese_sentences(text):
    sentences = []
    current_sentence = []
    i = 0
    n = len(text)
    
    # 状态跟踪
    in_quote = False  # 是否在引号中
    quote_chars = {'“': '”', '‘': '’'}  # 对应的引号匹配
    
    while i < n:
        char = text[i]
        
        # 处理引号状态
        if char in quote_chars:
            # 遇到开引号，进入引号状态
            current_sentence.append(char)
            expected_end_quote = quote_chars[char]
            i += 1
            # 寻找对应的闭引号
            while i < n and text[i] != expected_end_quote:
                current_sentence.append(text[i])
                i += 1
            if i < n:
                current_sentence.append(text[i])  # 添加闭引号
                i += 1
            continue
                
        # 处理省略号（六连点）
        if i <= n-5 and text[i:i+6] == '......':
            current_sentence.append('......')
            if not in_quote:
                sentences.append(''.join(current_sentence).strip())
                current_sentence = []
            i += 6
            continue
            
        # 处理中文省略号（三连顿号）
        if i <= n-2 and text[i] == '⋯' and text[i+1] == '⋯' and text[i+2] == '⋯':
            current_sentence.append('⋯⋯⋯')
            if not in_quote:
                sentences.append(''.join(current_sentence).strip())
                current_sentence = []
            i += 3
            continue
            
        # 处理普通结束符
        if char in {'。', '！', '？', '…'}:
            current_sentence.append(char)
            if not in_quote:
                sentences.append(''.join(current_sentence).strip())
                current_sentence = []
            i += 1
            continue
            
        # 默认情况：积累字符
        current_sentence.append(char)
        i += 1

    # 处理最后未完结的句子
    if current_sentence:
        sentences.append(''.join(current_sentence).strip())

    return [s for s in sentences if s]

权重调优

接下来，把两种方法搜索出来的内容，通过设定一个最小的rank数值，过滤掉无用的内容，再把两份内容结合起来，调整权重，最终找到最适合的几篇内容。

在文章切片以后，我选择的是通过切片内容，找到原文全部内容，整体喂给LLM。毕竟现在LLM的接口，入参的token数量大的已经到了1M，完全可以支持几万字，成本可以接受的情况下，为什么不充分利用呢？

PROMPT工程

最后一步就是调用LLM接口啦，这里主要是对prompt进行调优，特别是告诉LLM他的角色，用户的历史会话记录，以及刚才搜索出来的相关文档内容，设置合适的温度值，以及对输出内容的要求。这里有两个注意的地方，一个是我发现平台提供的LLM接口，即使使用同一个版本，也要定期检查它的输出内容，他一定几率会变得不可控，要及时调整prompt的写法；另一个是有些接口在设置角色、书写prompt时，英文的效果比中文好。

希望我的记录，对大家有所帮助。

其实是测试一下Stable Diffusion XL 支持的style，生成了好多好多，各式各样的小猫，废话不说，直接上图：

吴恩达在Coursera上又推出了一个新课程《Generative AI with Large Language Models》，学完以后发现非常不错，有系统性，又有实操性，所以把内容摘要在这里。

先来个大框架图，看一下搭建一个基于大模型的应用，怎么一步一步来做：

主要分为下面四个大步骤：确定场景、选择合适的预训练模型、基于应用场景对模型调优、最后模型部署和应用集成

1，确定场景：

目前AI应用场景主要包括：内容生成（文本、图片、音视频）、翻译、文本摘要和提取等等

2，选择合适的模型：

LLM大体上分成三类：

Encoder Only Models 又叫 Autoencodind Models，能利用上下文来生成缺失的内容，适合用于文本的情感分类、实体提取和单词分类。这类模型的代表就是BERT

Encoder Only Models 又叫 Autoregressive Models，它仅根据上文来生成后续内容，适合用于内容生成，这类模型的代表就是OpenAI的GPT、谷歌的BARD和Meta的LLaMA

Encoder Decoder Models 又叫 Sequence-to-Sequence Modles，它不直接对应上下文的词语，而是使用向量来预测内容，适合语言翻译、内容提取和问答，这类模型的代表包括T5和BART

关于transformers的更详细介绍，这里就不详细展开了。可以直接参考：Attention Is All You Need

根据要完成的任务，选择适合的大模型，还要注意模型的参数和精度，参数越多、精度越高，模型一般能力越强，但是成本成本也高，这里就需要平衡。

3，模型调优：

接下来是怎么把通用模型做优化，使得在特定场景下能够有更好的表现

3.1 Prompt engineering：

第一类方法不改变模型，也不需要训练，仅仅通过prompt里给模型一些提示，来改进模型的输出，所以也叫也叫In-context learning，根据是否在prompt中给输入输出举例，和举例的数量，分为下面几种：

zero shot：通过prompt本身的一些结构，让模型输出更好的结果，比如：

one shot：通过在prompt中给个例子，来让模型输出更好的结果，比如：

few shot：类似的，在prompt中给出多个例子（比如5-6个），来让模型输出更好的结果：

注意：当模型较小时，prompt工程的效果不会特别好

3.2 Prompt tuning and fine-tuning

第二类方法，用专门的数据对预训练好的通用大模型进行调优，一般针对某个任务，专门提供一些样本，来对模型进行训练，针对单个任务进行训练，就是通过标注好的数据集对模型进行进一步调优

但是这样做会遇到一个问题，就是Catastrophic forgetting，也就是模型会忘记以前的训练结果，针对其他任务，输出结果会变差，有两种方法来避免：

一种是使用多任务进行调优，叫Multi-task instruction fine-tuning

就是用多种任务的数据，对模型进行训练，增强模型在不同情况下的表现。

另一种方法就是Parameter efficient fine-tuning (PEFT)，通过冻结大部分参数，只训练小部分参数来进行：

或者冻结整个模型的参数，然后在原有模型外增加额外参数来训练：

来对模型进行调优。

这里常见的算法就是Low-Rank Adaptation of Large Language Models (LoRA)，原理如下：

几种方法的效果对比如下：

可以看到，当模型数量足够大时，几种方法的效果预期差距很小。

3.3 Reinforcement learning from human feedback (RLHF)

大模型的结果，经常会碰到输出的内容毫无意义，或者有攻击性、偏见等等情况，这时候就要再用人工标注的数据，进行一次强化学习。

首先要准备一批人工标注的数据，一般情况下，这样的数据量可能不够，不适合直接用于训练大模型，那就要用这批数据训练一个奖励模型，然后再用这个奖励模型，来对大模型进行训练：

这里常用的强化学习算法就叫：Proximal Policy Optimization Algorithms (PPO)，这里就不详细展开了。但是，我们会遇到一个问题，模型会过分关注奖励，导致结果不合理，这种情况叫reward hacking。要避免reward hacking，可以计算原始模型和强化学习以后模型输出的结果，把差异特别大的结果，进行额外惩罚。

同样，我们可以用前面类似PEFT的方法，来替代直接更新所有模型参数，如下所示：

这里引出一个特别重要的概念，就是AI大模型的监管，如何让AI大模型是对人类有帮助而且无害的，可以参考：Constitutional AI: Harmlessness from AI Feedback

3.4 模型评价

针对应用场景，应该有自己的模型评价指标，课程里介绍了几种常用的，包括ROUGE、BLEU等，和常用的Benchmark：SuperGLUE、HELM

4. 模型部署：

4.1 模型压缩

符合要求的模型，在正式部署前，可以做模型压缩，一般有三条路线：

Distillation是用大模型来训练一个相对较小的模型

Quantization是降低大模型参数的精度，比如把16bit的浮点数降为8bit的整数

Pruning是去掉大模型中等于0或者非常接近0的参数

下面是几种对模型进行优化的方法对比：

4.2 模型与外部数据的连接

大模型一般基于公开数据、一定时间以前数据进行训练的，为了让大模型能够使用应用自身的数据、最新的数据，我们要用Retrieval augmented generation (RAG)技术和Agent技术。

RAG是通过检索外部数据、并向量化，然后传给大模型，来让大模型能够使用特定的数据

Agent就是通过调用API来使用其他应用的数据或者服务：

课程中举了Program-aided language (PAL) models和

4.3 基于大模型的应用架构

最后来看一下，一个基于大模型的应用的整体架构：

最底层是基础架构，包括训练和模型调优；上面一般是应用自己的特定数据源、模型本身和用户反馈数据；再上面是大模型工具集合，这个往往是开发工程量最大的部分；再上面就是应用自身的服务，比如网站、App、API等等。

自从chatGPT火爆兴起，到Midjourney图片生成，AI又一次站到了风口浪尖，而且来势汹汹，大有颠覆行业之势。那么到底大模型AI距离人类，差距在哪些地方？

要说差距，先看大模型AI到底能做到些什么？大模型，无论是chatGPT还是谷歌的Bard，原理都是用足够多的参数（数千亿，上万亿），和足够多的数据（号称人类全部知识）去训练模型。所谓AI模型，就是去模拟人类的输入-输出，参数量越大，模拟得越准确。

再来看人类认知能力是怎样的。从哲学的“认识论”上来说，人类认识能力，无外乎先天知识和经验知识。先天知识是与生俱来的认知能力，而经验知识是在客观环境中，通过先天知识获得的反馈。

先天知识对应到GPT就是预训练模型，只是这个预训练模型特别牛逼，是生命出现在地球上以来，大概35亿年的训练过程，加上好几次非常特别的参数变化：大洪水、大火山、陨石撞地球等等。训练的结果就是DNA，有效性息量大概也就3.2G（当然也可能我们对遗传物质的认知不够），而GPT3的预训练模型大小已经有几百G。其存储效率以及对应的能耗差别巨大。

经验知识，更接近用预训练模型针对特殊场景的训练，人类这个过程就是从出生到发育完成的过程，经历十多年，并终身不断更新。不同的生活背景、教育背景、把每个人都训练成了不一样的模型。

然而人类作为一个种群，还有跟现在大模型AI不一样的地方，就是种群内大量的模型，进行自然选择。OpenAI再厉害，也就训练出来了一个模型（或者几个），但是人类同时训练着几十亿个模型，历史上曾经出现过上千亿的人类，也就是训练了上千亿的模型，并通过自然选择，传承了最合适的一部分。单体模型和群体模型，有着根本性的区别，也就是社会性。人类作为一个独立个体，失去了社会性，只剩下动物性，其认知能力可能比现在的大模型AI更弱。我觉得这也是目前大模型AI的能力天花板。

人类社会性有哪些体现？简单说，包括：道德、阶级、自觉、利他、服从、依赖等等，这些确实是目前AI与人类最大的区别。

所以，AI目前的发展道路，指向的还是工具、更高级的工具，更智能的工具，而非人类的替代。就像工业化进程，并没有让人类被机器替代，反而让人类的生产能力获得空前的提升，物资空前的丰富。（本文配图来自百度文心一言）

AI真正要变成人类，先解决模型的能效问题，再通过大量的模型之间的合作与竞争，建立社会性，才有可能。