【设计模式】使用python 实践框架设计

1. 单一职责原则（SRP）：一个类应该只有一个职责，意味着该类只应该有一个引起变化的原因。这使得代码更易于维护和理解。

2. 开放封闭原则（OCP）：软件实体（类、模块、函数等）应该对扩展开放，对修改封闭。这意味着可以通过添加新代码来扩展功能，而不是修改现有代码。

3. 里氏替换原则（LSP）：子类对象应该能够替换父类对象而不影响程序的正确性。这要求子类必须完全实现父类的行为。

4. 接口隔离原则（ISP）：不应强迫一个类依赖于它不使用的接口。多个特定客户端接口要好于一个通用接口。

5. 依赖反转原则（DIP）：高层模块不应该依赖于低层模块，两者都应该依赖于抽象。抽象不应该依赖于细节，细节应该依赖于抽象。

6. 合成复用原则（CRP）：优先使用对象组合而不是继承来达到复用的目的。组合可以提供更灵活的解决方案，并且避免了继承带来的复杂性。

结合机器学习中大模型微调的数据处理场景，尝试梳理学习以下设计模式：

• 模板方法模式 (Template Method)：在方法中定义算法框架。
• 策略模式 (Strategy)：定义一系列算法，使算法可以互换，提高系统的灵活性。
• 组合模式 (Composite)：部分-整体层次关系，简化对复杂树形结构的操作。
• 适配器模式 (Adapter)：将一个接口转换为另一个接口，解决不兼容接口的问题，增加系统的可复用性。
• 建造者模式 (Builder)：构建复杂对象的步骤，逐步配置对象，便于管理和扩展。
• 工厂方法模式 (Factory Method)：需要创建对象时，子类决定具体类，增加新形状时，不需要修改现有代码，只需实现新类。

#coding:utf8
import pandas as pd
from abc import ABC, abstractmethod
import jsonclass Trainset:def __init__(self):self.raw_data_path = Noneself.feature_column_list = Noneself.label_strategy = Noneself.trainset_ratio = Noneself.train_data = Noneself.test_data = Noneself.oversample = Noneself.data_format = Nonedef __str__(self):return f"raw data path: {self.raw_data_path}\nfeature_column_list: {self.feature_column_list}\nlabel_strategy: {self.label_strategy}\n" \f"trainset_ratio: {self.trainset_ratio}\ntrain_df: {len(self.train_data)}\ntest_df: {len(self.test_data)}\nsample:\n{self.train_data.iloc[7]}"def save_train_data(self, path):with open(path, "w", encoding="utf8") as fout:for i in range(len(self.train_data)):fout.write(self.train_data.iloc[i]+"\n")def save_test_data(self, path):with open(path, "w", encoding="utf8") as fout:for i in range(len(self.test_data)):fout.write(self.test_data.iloc[i] + "\n")
class LabelStrategy(ABC):@staticmethod@abstractmethoddef gen_label(row):passclass ImportantType1(LabelStrategy):@staticmethoddef gen_label(row):if row['是否重要新闻标签'] == 1 and row['判断条件一'] == 'Y':return 1else:return 0class ImportantType2(LabelStrategy):@staticmethoddef gen_label(row):if row['是否重要新闻标签'] == 1 and row['判断条件二'] == 'Y':return 1else:return 0class ImportantType3(LabelStrategy):@staticmethoddef gen_label(row):if row['是否重要新闻标签'] == 1 and row['判断条件三'] == 'Y':return 1else:return 0class ImportantType4(LabelStrategy):@staticmethoddef gen_label(row):if row['是否重要新闻标签'] == 1 and row['判断条件四'] == 'Y':return 1else:return 0class DataFormat(ABC):@staticmethod@abstractmethoddef transform(row):passclass BertDataFormat(DataFormat):@staticmethoddef transform(row):return str(row['label']) + "\t" + row["feature"]class QwenDataFormat(DataFormat):prompt = "请判断以下新闻会不会对对应股票价格造成重大负面影响，造成股价异常下跌？会导致股价大幅下跌输出1，不会输出0。新闻为："@classmethod# classmethod和staticmethod的共同的是可以不实例化类就调用类内方法，区别是classmethod可以通过cls使用类内变量，而staticmethod无法调用类内变量def transform(cls, row):return json.dumps({"type": "chatml", "message":[{"role": "user", "content": cls.prompt+row['feature']},{"role": "assistant", "content": str(row['label'])}],"source": "self-made"}, ensure_ascii=False)class TrainsetBuilder:def __init__(self):self.trainset = Trainset()self.train_df = Noneself.test_df = Noneself.data_format_dict = {'bert': BertDataFormat,'qwen': QwenDataFormat}def set_data_path(self, raw_data_path):self.trainset.raw_data_path = raw_data_pathreturn selfdef set_feature(self, feature_column_list):self.trainset.feature_column_list = feature_column_listreturn selfdef set_label_strategy(self, label_strategy):self.trainset.label_strategy = label_strategyreturn selfdef set_trainset_ratio(self, ratio):self.trainset.trainset_ratio = ratioreturn selfdef set_data_format(self, data_format):self.trainset.data_format = data_formatreturn selfdef set_oversample(self, oversample=True):self.trainset.oversample = oversamplereturn selfdef balance_label(self):pos_df = self.train_df[self.train_df['label'].isin([1])]neg_df = self.train_df[self.train_df['label'].isin([0])]if len(neg_df) > 1.5 * len(pos_df):oversampel_ratio = int(len(neg_df)/len(pos_df))print(f"pos:{len(pos_df)}, neg:{len(neg_df)}, oversampel_ratio:{oversampel_ratio}")pos_df = pd.concat([pos_df] * oversampel_ratio, ignore_index=True)elif len(pos_df) > 1.5 * len(neg_df):oversampel_ratio = int(len(pos_df) / len(neg_df))print(f"pos:{len(pos_df)}, neg:{len(neg_df)}, oversampel_ratio:{oversampel_ratio}")neg_df = pd.concat([neg_df] * oversampel_ratio, ignore_index=True)train_df = pd.concat([pos_df, neg_df])self.train_df = train_df.sample(frac=1, random_state=87).reset_index(drop=True)def build(self):data_df = pd.read_csv(self.trainset.raw_data_path, encoding="gbk")data_df['feature'] = data_df.apply(lambda row: ",".join([row[i] for i in self.trainset.feature_column_list]),  axis=1)data_df['label'] = data_df.apply(lambda row: self.trainset.label_strategy.gen_label(row),  axis=1)data_df = data_df[['feature', 'label']]data_df = data_df.sample(frac=1, random_state=42).reset_index(drop=True)self.train_df = data_df.head(int(len(data_df) * self.trainset.trainset_ratio))self.test_df = data_df.tail(len(data_df) - len(self.train_df))if self.trainset.oversample:self.balance_label()self.trainset.train_data = self.train_df.apply(lambda row: self.data_format_dict.get(self.trainset.data_format, BertDataFormat).transform(row), axis=1)print(type(self.trainset.train_data))print(self.trainset.train_data)self.trainset.test_data = self.test_df.apply(lambda row: self.data_format_dict.get(self.trainset.data_format, BertDataFormat).transform(row), axis=1)return self.trainsetif __name__ == "__main__":builder = TrainsetBuilder()trainset = (builder.set_data_path("./raw_data/outputresult.csv").set_feature(['新闻标题']).set_label_strategy(ImportantType4) #ImportantType1, ImportantType2, ImportantType3, ImportantType4.set_trainset_ratio(0.8).set_oversample(True).set_data_format('bert') #bert, qwen.build())print(trainset)output_dir = "./data/"trainset.save_train_data(output_dir + "bert_train.tsv")trainset.save_test_data(output_dir + "bert_test.tsv")