12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108210921102111211221132114211521162117211821192120212121222123212421252126212721282129213021312132213321342135213621372138213921402141214221432144214521462147214821492150215121522153215421552156215721582159216021612162216321642165216621672168216921702171217221732174217521762177217821792180218121822183218421852186218721882189219021912192219321942195219621972198219922002201220222032204220522062207220822092210221122122213221422152216221722182219222022212222222322242225222622272228222922302231223222332234223522362237223822392240224122422243224422452246224722482249225022512252225322542255225622572258225922602261226222632264226522662267226822692270227122722273227422752276227722782279228022812282228322842285228622872288228922902291229222932294229522962297229822992300230123022303230423052306230723082309231023112312231323142315231623172318231923202321232223232324232523262327232823292330233123322333233423352336233723382339234023412342234323442345234623472348234923502351235223532354235523562357235823592360236123622363236423652366236723682369237023712372237323742375237623772378237923802381238223832384238523862387238823892390239123922393239423952396239723982399240024012402240324042405240624072408240924102411241224132414241524162417241824192420242124222423242424252426242724282429243024312432243324342435243624372438243924402441244224432444244524462447244824492450245124522453245424552456245724582459246024612462246324642465246624672468246924702471247224732474247524762477247824792480248124822483248424852486248724882489249024912492249324942495249624972498249925002501250225032504250525062507250825092510251125122513251425152516251725182519252025212522252325242525252625272528252925302531253225332534253525362537253825392540254125422543254425452546254725482549255025512552255325542555255625572558255925602561256225632564256525662567256825692570257125722573257425752576257725782579258025812582258325842585258625872588258925902591259225932594259525962597259825992600260126022603260426052606260726082609261026112612261326142615261626172618261926202621262226232624262526262627262826292630263126322633263426352636263726382639264026412642264326442645264626472648264926502651265226532654265526562657265826592660266126622663266426652666266726682669267026712672267326742675267626772678267926802681268226832684268526862687268826892690269126922693269426952696269726982699270027012702270327042705270627072708270927102711271227132714271527162717271827192720272127222723272427252726272727282729273027312732273327342735273627372738273927402741274227432744274527462747274827492750275127522753275427552756275727582759276027612762276327642765276627672768276927702771277227732774277527762777277827792780278127822783278427852786278727882789279027912792279327942795279627972798279928002801280228032804280528062807280828092810281128122813281428152816281728182819282028212822282328242825282628272828282928302831283228332834283528362837283828392840284128422843284428452846284728482849285028512852285328542855285628572858285928602861286228632864286528662867286828692870287128722873287428752876287728782879288028812882288328842885288628872888288928902891289228932894289528962897289828992900290129022903290429052906290729082909291029112912291329142915291629172918291929202921292229232924292529262927292829292930293129322933293429352936293729382939294029412942294329442945294629472948294929502951295229532954295529562957295829592960296129622963296429652966296729682969297029712972297329742975297629772978297929802981298229832984298529862987298829892990299129922993299429952996299729982999300030013002300330043005300630073008300930103011301230133014301530163017301830193020302130223023302430253026302730283029303030313032303330343035303630373038303930403041304230433044304530463047304830493050305130523053305430553056305730583059306030613062306330643065306630673068306930703071307230733074307530763077307830793080308130823083308430853086308730883089309030913092309330943095309630973098309931003101310231033104310531063107310831093110311131123113311431153116311731183119312031213122312331243125312631273128312931303131313231333134313531363137313831393140314131423143314431453146314731483149315031513152315331543155315631573158315931603161316231633164316531663167316831693170317131723173317431753176317731783179318031813182318331843185318631873188318931903191319231933194319531963197319831993200320132023203320432053206320732083209321032113212321332143215321632173218321932203221322232233224322532263227322832293230323132323233323432353236323732383239324032413242324332443245324632473248324932503251325232533254325532563257325832593260326132623263326432653266326732683269327032713272327332743275327632773278327932803281328232833284328532863287328832893290329132923293329432953296329732983299330033013302330333043305330633073308330933103311331233133314331533163317331833193320332133223323332433253326332733283329333033313332333333343335333633373338333933403341334233433344334533463347334833493350335133523353335433553356335733583359336033613362336333643365336633673368336933703371337233733374337533763377337833793380338133823383338433853386338733883389339033913392339333943395339633973398339934003401340234033404340534063407340834093410341134123413341434153416341734183419342034213422342334243425342634273428342934303431343234333434343534363437343834393440344134423443344434453446344734483449345034513452345334543455345634573458345934603461346234633464346534663467346834693470347134723473347434753476347734783479348034813482348334843485348634873488348934903491349234933494349534963497349834993500350135023503350435053506350735083509351035113512351335143515351635173518351935203521352235233524352535263527352835293530353135323533353435353536353735383539354035413542354335443545354635473548354935503551355235533554355535563557355835593560356135623563356435653566356735683569357035713572357335743575357635773578357935803581358235833584358535863587358835893590359135923593359435953596359735983599360036013602360336043605360636073608360936103611361236133614361536163617361836193620362136223623362436253626362736283629363036313632363336343635363636373638363936403641364236433644364536463647364836493650365136523653365436553656365736583659366036613662366336643665366636673668366936703671367236733674367536763677367836793680368136823683368436853686368736883689369036913692369336943695369636973698369937003701370237033704370537063707370837093710371137123713371437153716371737183719372037213722372337243725372637273728372937303731373237333734373537363737373837393740374137423743374437453746374737483749375037513752375337543755375637573758375937603761376237633764376537663767376837693770377137723773377437753776377737783779378037813782378337843785378637873788378937903791379237933794379537963797379837993800380138023803380438053806380738083809381038113812381338143815381638173818381938203821382238233824382538263827382838293830383138323833383438353836383738383839384038413842384338443845384638473848384938503851385238533854385538563857385838593860386138623863386438653866386738683869387038713872387338743875387638773878387938803881388238833884388538863887388838893890389138923893389438953896389738983899390039013902390339043905390639073908390939103911391239133914391539163917391839193920392139223923392439253926392739283929393039313932393339343935393639373938393939403941394239433944394539463947394839493950395139523953395439553956395739583959396039613962396339643965396639673968396939703971397239733974397539763977397839793980398139823983398439853986398739883989399039913992399339943995399639973998399940004001400240034004400540064007400840094010401140124013401440154016401740184019402040214022402340244025402640274028402940304031403240334034403540364037403840394040404140424043404440454046404740484049405040514052405340544055405640574058405940604061406240634064406540664067406840694070407140724073407440754076407740784079408040814082408340844085408640874088408940904091409240934094409540964097409840994100410141024103410441054106410741084109411041114112411341144115411641174118411941204121412241234124412541264127412841294130413141324133413441354136413741384139414041414142414341444145414641474148414941504151415241534154415541564157415841594160416141624163416441654166416741684169417041714172417341744175417641774178417941804181418241834184418541864187418841894190419141924193419441954196419741984199420042014202420342044205420642074208420942104211421242134214421542164217421842194220422142224223422442254226422742284229423042314232423342344235423642374238423942404241424242434244424542464247424842494250425142524253425442554256425742584259426042614262426342644265426642674268426942704271427242734274427542764277427842794280428142824283428442854286428742884289429042914292429342944295429642974298429943004301430243034304430543064307430843094310431143124313431443154316431743184319432043214322432343244325432643274328432943304331433243334334433543364337433843394340434143424343434443454346434743484349435043514352435343544355435643574358435943604361436243634364436543664367436843694370437143724373437443754376437743784379438043814382438343844385438643874388438943904391439243934394439543964397439843994400440144024403440444054406440744084409441044114412441344144415441644174418441944204421442244234424442544264427442844294430443144324433443444354436443744384439444044414442444344444445444644474448444944504451445244534454445544564457445844594460446144624463446444654466446744684469447044714472447344744475447644774478447944804481448244834484448544864487448844894490449144924493449444954496449744984499450045014502450345044505450645074508450945104511451245134514451545164517451845194520452145224523452445254526452745284529453045314532453345344535453645374538453945404541454245434544454545464547454845494550455145524553455445554556455745584559456045614562456345644565456645674568456945704571457245734574457545764577457845794580458145824583458445854586458745884589459045914592459345944595459645974598459946004601460246034604460546064607460846094610461146124613 |
- from scipy.fftpack import fft, ifft, ifftn, fftn # 快速傅里叶变换
- from sklearn.svm import SVC, SVR # SVC是svm分类,SVR是svm回归
- from pyecharts.components import Table as Table_Fisrt # 绘制表格
- from scipy import optimize
- from sklearn.cluster import KMeans, AgglomerativeClustering, DBSCAN
- from sklearn.manifold import TSNE
- from sklearn.neural_network import MLPClassifier, MLPRegressor
- from sklearn.discriminant_analysis import LinearDiscriminantAnalysis as LDA
- from sklearn.decomposition import PCA, IncrementalPCA, KernelPCA, NMF
- from sklearn.impute import SimpleImputer
- from sklearn.preprocessing import *
- from sklearn.feature_selection import *
- from sklearn.metrics import *
- from sklearn.ensemble import (
- RandomForestClassifier,
- RandomForestRegressor,
- GradientBoostingClassifier,
- GradientBoostingRegressor)
- from sklearn.tree import DecisionTreeClassifier, DecisionTreeRegressor, export_graphviz
- from sklearn.neighbors import KNeighborsClassifier, KNeighborsRegressor
- from sklearn.linear_model import *
- from sklearn.model_selection import train_test_split
- import re
- import numpy as np
- from pandas import DataFrame, read_csv
- import matplotlib.pyplot as plt
- from scipy.cluster.hierarchy import dendrogram, ward
- from pyecharts.options.series_options import JsCode
- from pyecharts.charts import Tab as tab_First
- from pyecharts.charts import *
- from random import randint
- from pyecharts import options as opts
- from pyecharts.components import Image
- from os.path import split as path_split
- from os.path import exists, basename, splitext
- from os import mkdir, getcwd
- import tarfile
- import pickle
- import joblib
- from pyecharts.globals import CurrentConfig
- CurrentConfig.ONLINE_HOST = f"{getcwd()}/assets/"
- # 设置
- np.set_printoptions(threshold=np.inf)
- global_Set = dict(
- toolbox_opts=opts.ToolboxOpts(
- is_show=True), legend_opts=opts.LegendOpts(
- pos_bottom='3%', type_='scroll'))
- global_Leg = dict(
- toolbox_opts=opts.ToolboxOpts(
- is_show=True), legend_opts=opts.LegendOpts(
- is_show=False))
- Label_Set = dict(label_opts=opts.LabelOpts(is_show=False))
- More_Global = False # 是否使用全部特征绘图
- All_Global = True # 是否导出charts
- CSV_Global = True # 是否导出CSV
- CLF_Global = True # 是否导出模型
- TAR_Global = True # 是否打包tar
- NEW_Global = True # 是否新建目录
- class Tab(tab_First):
- def __init__(self, *args, **kwargs):
- super(Tab, self).__init__(*args, **kwargs)
- self.element = {} # 记录tab组成元素 name:charts
- def add(self, chart, tab_name):
- self.element[tab_name] = chart
- return super(Tab, self).add(chart, tab_name)
- def render(
- self,
- path: str = "render.html",
- template_name: str = "simple_tab.html",
- *args,
- **kwargs,
- ) -> str:
- if All_Global:
- Dic = path_split(path)[0]
- for i in self.element:
- self.element[i].render(Dic + '/' + i + '.html')
- return super(Tab, self).render(path, template_name, *args, **kwargs)
- class Table(Table_Fisrt):
- def __init__(self, *args, **kwargs):
- super(Table, self).__init__(*args, **kwargs)
- self.HEADERS = []
- self.ROWS = [[]]
- def add(self, headers, rows, attributes=None):
- if len(rows) == 1:
- new_headers = ['数据类型', '数据']
- new_rows = list(zip(headers, rows[0]))
- self.HEADERS = new_headers
- self.ROWS = new_rows
- return super().add(new_headers, new_rows, attributes)
- else:
- self.HEADERS = headers
- self.ROWS = rows
- return super().add(headers, rows, attributes)
- def render(self, path="render.html", *args, **kwargs,) -> str:
- if CSV_Global:
- Dic, name = path_split(path)
- name = splitext(name)[0]
- try:
- DataFrame(
- self.ROWS,
- columns=self.HEADERS).to_csv(
- Dic + '/' + name + '.csv')
- except BaseException:
- pass
- return super().render(path, *args, **kwargs)
- def make_list(first, end, num=35):
- n = num / (end - first)
- if n == 0:
- n = 1
- re = []
- n_first = first * n
- n_end = end * n
- while n_first <= n_end:
- cul = n_first / n
- re.append(round(cul, 2))
- n_first += 1
- return re
- def list_filter(list_, num=70):
- # 假设列表已经不重复
- if len(list_) <= num:
- return list_
- n = int(num / len(list_))
- re = list_[::n]
- return re
- def Prediction_boundary(x_range, x_means, Predict_Func, Type): # 绘制回归型x-x热力图
- # r是绘图大小列表,x_means是其余值,Predict_Func是预测方法回调
- # a-特征x,b-特征x-1,c-其他特征
- o_cList = []
- if len(x_means) == 1:
- return o_cList
- for i in range(len(x_means)):
- for j in range(len(x_means)):
- if j <= i:
- continue
- n_ra = x_range[j]
- Type_ra = Type[j]
- n_rb = x_range[i]
- Type_rb = Type[i]
- if Type_ra == 1:
- ra = make_list(n_ra[0], n_ra[1], 70)
- else:
- ra = list_filter(n_ra) # 可以接受最大为70
- if Type_rb == 1:
- rb = make_list(n_rb[0], n_rb[1], 35)
- else:
- rb = list_filter(n_rb) # 可以接受最大为70
- a = np.array([i for i in ra for _ in rb]).T
- b = np.array([i for _ in ra for i in rb]).T
- data = np.array([x_means for _ in ra for i in rb])
- data[:, j] = a
- data[:, i] = b
- y_data = Predict_Func(data)[0].tolist()
- value = [[float(a[i]), float(b[i]), y_data[i]]
- for i in range(len(a))]
- c = (HeatMap()
- .add_xaxis(np.unique(a))
- # value的第一个数值是x
- .add_yaxis(f'数据', np.unique(b), value, **Label_Set)
- .set_global_opts(title_opts=opts.TitleOpts(title='预测热力图'), **global_Leg,
- yaxis_opts=opts.AxisOpts(
- is_scale=True, type_='category'), # 'category'
- xaxis_opts=opts.AxisOpts(is_scale=True, type_='category'),
- visualmap_opts=opts.VisualMapOpts(is_show=True, max_=int(max(y_data)) + 1, min_=int(min(y_data)),
- pos_right='3%')) # 显示
- )
- o_cList.append(c)
- return o_cList
- def Prediction_boundary_More(x_range, x_means, Predict_Func, Type): # 绘制回归型x-x热力图
- # r是绘图大小列表,x_means是其余值,Predict_Func是预测方法回调
- # a-特征x,b-特征x-1,c-其他特征
- o_cList = []
- if len(x_means) == 1:
- return o_cList
- for i in range(len(x_means)):
- if i == 0:
- continue
- n_ra = x_range[i - 1]
- Type_ra = Type[i - 1]
- n_rb = x_range[i]
- Type_rb = Type[i]
- if Type_ra == 1:
- ra = make_list(n_ra[0], n_ra[1], 70)
- else:
- ra = list_filter(n_ra) # 可以接受最大为70
- if Type_rb == 1:
- rb = make_list(n_rb[0], n_rb[1], 35)
- else:
- rb = list_filter(n_rb) # 可以接受最大为70
- a = np.array([i for i in ra for _ in rb]).T
- b = np.array([i for _ in ra for i in rb]).T
- data = np.array([x_means for _ in ra for i in rb])
- data[:, i - 1] = a
- data[:, i] = b
- y_data = Predict_Func(data)[0].tolist()
- value = [[float(a[i]), float(b[i]), y_data[i]] for i in range(len(a))]
- c = (HeatMap()
- .add_xaxis(np.unique(a))
- # value的第一个数值是x
- .add_yaxis(f'数据', np.unique(b), value, **Label_Set)
- .set_global_opts(title_opts=opts.TitleOpts(title='预测热力图'), **global_Leg,
- yaxis_opts=opts.AxisOpts(
- is_scale=True, type_='category'), # 'category'
- xaxis_opts=opts.AxisOpts(is_scale=True, type_='category'),
- visualmap_opts=opts.VisualMapOpts(is_show=True, max_=int(max(y_data)) + 1, min_=int(min(y_data)),
- pos_right='3%')) # 显示
- )
- o_cList.append(c)
- return o_cList
- def Decision_boundary(x_range, x_means, Predict_Func, class_, Type, nono=False): # 绘制分类型预测图x-x热力图
- # r是绘图大小列表,x_means是其余值,Predict_Func是预测方法回调,class_是分类,add_o是可以合成的图
- # a-特征x,b-特征x-1,c-其他特征
- # 规定,i-1是x轴,a是x轴,x_1是x轴
- class_dict = dict(zip(class_, [i for i in range(len(class_))]))
- if not nono:
- v_dict = [{'min': -1.5, 'max': -0.5, 'label': '未知'}] # 分段显示
- else:
- v_dict = []
- for i in class_dict:
- v_dict.append(
- {'min': class_dict[i] - 0.5, 'max': class_dict[i] + 0.5, 'label': str(i)})
- o_cList = []
- if len(x_means) == 1:
- n_ra = x_range[0]
- if Type[0] == 1:
- ra = make_list(n_ra[0], n_ra[1], 70)
- else:
- ra = n_ra
- a = np.array([i for i in ra]).reshape(-1, 1)
- y_data = Predict_Func(a)[0].tolist()
- value = [[0, float(a[i]), class_dict.get(y_data[i], -1)]
- for i in range(len(a))]
- c = (HeatMap()
- .add_xaxis(['None'])
- # value的第一个数值是x
- .add_yaxis(f'数据', np.unique(a), value, **Label_Set)
- .set_global_opts(title_opts=opts.TitleOpts(title='预测热力图'), **global_Leg,
- yaxis_opts=opts.AxisOpts(
- is_scale=True, type_='category'), # 'category'
- xaxis_opts=opts.AxisOpts(is_scale=True, type_='category'),
- visualmap_opts=opts.VisualMapOpts(is_show=True, max_=max(class_dict.values()),
- min_=-1,
- is_piecewise=True, pieces=v_dict,
- orient='horizontal', pos_bottom='3%'))
- )
- o_cList.append(c)
- return o_cList
- # 如果x_means长度不等于1则执行下面
- for i in range(len(x_means)):
- if i == 0:
- continue
- n_ra = x_range[i - 1]
- Type_ra = Type[i - 1]
- n_rb = x_range[i]
- Type_rb = Type[i]
- if Type_ra == 1:
- ra = make_list(n_ra[0], n_ra[1], 70)
- else:
- ra = n_ra
- if Type_rb == 1:
- rb = make_list(n_rb[0], n_rb[1], 35)
- else:
- rb = n_rb
- a = np.array([i for i in ra for _ in rb]).T
- b = np.array([i for _ in ra for i in rb]).T
- data = np.array([x_means for _ in ra for i in rb])
- data[:, i - 1] = a
- data[:, i] = b
- y_data = Predict_Func(data)[0].tolist()
- value = [[float(a[i]), float(b[i]), class_dict.get(y_data[i], -1)]
- for i in range(len(a))]
- c = (HeatMap()
- .add_xaxis(np.unique(a))
- # value的第一个数值是x
- .add_yaxis(f'数据', np.unique(b), value, **Label_Set)
- .set_global_opts(title_opts=opts.TitleOpts(title='预测热力图'), **global_Leg,
- yaxis_opts=opts.AxisOpts(
- is_scale=True, type_='category'), # 'category'
- xaxis_opts=opts.AxisOpts(is_scale=True, type_='category'),
- visualmap_opts=opts.VisualMapOpts(is_show=True, max_=max(class_dict.values()), min_=-1,
- is_piecewise=True, pieces=v_dict, orient='horizontal', pos_bottom='3%'))
- )
- o_cList.append(c)
- return o_cList
- # 绘制分类型预测图x-x热力图
- def Decision_boundary_More(
- x_range,
- x_means,
- Predict_Func,
- class_,
- Type,
- nono=False):
- # r是绘图大小列表,x_means是其余值,Predict_Func是预测方法回调,class_是分类,add_o是可以合成的图
- # a-特征x,b-特征x-1,c-其他特征
- # 规定,i-1是x轴,a是x轴,x_1是x轴
- class_dict = dict(zip(class_, [i for i in range(len(class_))]))
- if not nono:
- v_dict = [{'min': -1.5, 'max': -0.5, 'label': '未知'}] # 分段显示
- else:
- v_dict = []
- for i in class_dict:
- v_dict.append(
- {'min': class_dict[i] - 0.5, 'max': class_dict[i] + 0.5, 'label': str(i)})
- o_cList = []
- if len(x_means) == 1:
- return Decision_boundary(
- x_range,
- x_means,
- Predict_Func,
- class_,
- Type,
- nono)
- # 如果x_means长度不等于1则执行下面
- for i in range(len(x_means)):
- for j in range(len(x_means)):
- if j <= i:
- continue
- n_ra = x_range[j]
- Type_ra = Type[j]
- n_rb = x_range[i]
- Type_rb = Type[i]
- if Type_ra == 1:
- ra = make_list(n_ra[0], n_ra[1], 70)
- else:
- ra = n_ra
- if Type_rb == 1:
- rb = make_list(n_rb[0], n_rb[1], 35)
- else:
- rb = n_rb
- a = np.array([i for i in ra for _ in rb]).T
- b = np.array([i for _ in ra for i in rb]).T
- data = np.array([x_means for _ in ra for i in rb])
- data[:, j] = a
- data[:, i] = b
- y_data = Predict_Func(data)[0].tolist()
- value = [[float(a[i]), float(b[i]), class_dict.get(
- y_data[i], -1)] for i in range(len(a))]
- c = (HeatMap()
- .add_xaxis(np.unique(a))
- # value的第一个数值是x
- .add_yaxis(f'数据', np.unique(b), value, **Label_Set)
- .set_global_opts(title_opts=opts.TitleOpts(title='预测热力图'), **global_Leg,
- yaxis_opts=opts.AxisOpts(
- is_scale=True, type_='category'), # 'category'
- xaxis_opts=opts.AxisOpts(is_scale=True, type_='category'),
- visualmap_opts=opts.VisualMapOpts(is_show=True, max_=max(class_dict.values()), min_=-1,
- is_piecewise=True, pieces=v_dict, orient='horizontal', pos_bottom='3%'))
- )
- o_cList.append(c)
- return o_cList
- def SeeTree(Dic):
- node_re = re.compile(r'^([0-9]+) \[label="(.+)"\] ;$') # 匹配节点正则表达式
- link_re = re.compile('^([0-9]+) -> ([0-9]+) (.*);$') # 匹配节点正则表达式
- node_Dict = {}
- link_list = []
- with open(Dic, 'r') as f: # 貌似必须分开w和r
- for i in f:
- try:
- get = re.findall(node_re, i)[0]
- if get[0] != '':
- try:
- v = float(get[0])
- except BaseException:
- v = 0
- node_Dict[get[0]] = {'name': get[1].replace(
- '\\n', '\n'), 'value': v, 'children': []}
- continue
- except BaseException:
- pass
- try:
- get = re.findall(link_re, i)[0]
- if get[0] != '' and get[1] != '':
- link_list.append((get[0], get[1]))
- except BaseException:
- pass
- father_list = [] # 已经有父亲的list
- for i in link_list:
- father = i[0] # 父节点
- son = i[1] # 子节点
- try:
- node_Dict[father]['children'].append(node_Dict[son])
- father_list.append(son)
- if int(son) == 0:
- print('F')
- except BaseException:
- pass
- father = list(set(node_Dict.keys()) - set(father_list))
- c = (
- Tree()
- .add("", [node_Dict[father[0]]], is_roam=True)
- .set_global_opts(title_opts=opts.TitleOpts(title="决策树可视化"),
- toolbox_opts=opts.ToolboxOpts(is_show=True))
- )
- return c
- def make_Tab(heard, row):
- return Table().add(headers=heard, rows=row)
- def scatter(w_heard, w):
- c = (Scatter() .add_xaxis(w_heard) .add_yaxis('',
- w,
- **Label_Set) .set_global_opts(title_opts=opts.TitleOpts(title='系数w散点图'),
- **global_Set))
- return c
- def bar(w_heard, w):
- c = (Bar() .add_xaxis(w_heard) .add_yaxis('',
- abs(w).tolist(),
- **Label_Set) .set_global_opts(title_opts=opts.TitleOpts(title='系数w柱状图'),
- **global_Set))
- return c
- # def line(w_sum,w,b):
- # x = np.arange(-5, 5, 1)
- # c = (
- # Line()
- # .add_xaxis(x.tolist())
- # .set_global_opts(title_opts=opts.TitleOpts(title=f"系数w曲线"), **global_Set)
- # )
- # for i in range(len(w)):
- # y = x * w[i] + b
- # c.add_yaxis(f"系数w[{i}]", y.tolist(), is_smooth=True, **Label_Set)
- # return c
- def see_Line(x_trainData, y_trainData, w, w_sum, b):
- y = y_trainData.tolist()
- x_data = x_trainData.transpose
- re = []
- for i in range(len(x_data)):
- x = x_data[i]
- p = int(x.max() - x.min()) / 5
- x_num = np.arange(x.min(), x.min() + p * 6, p) # 固定5个点,并且正好包括端点
- y_num = x_num * w[i] + (w[i] / w_sum) * b
- c = (
- line() .add_xaxis(
- x_num.tolist()) .add_yaxis(
- f"{i}预测曲线",
- y_num.tolist(),
- is_smooth=True,
- **Label_Set) .set_global_opts(
- title_opts=opts.TitleOpts(
- title=f"系数w曲线"),
- **global_Set))
- t = (
- Scatter() .add_xaxis(
- x.tolist()) .add_yaxis(
- f'{i}特征',
- y,
- **Label_Set) .set_global_opts(
- title_opts=opts.TitleOpts(
- title='类型划分图'),
- **global_Set))
- t.overlap(c)
- re.append(t)
- return re
- def get_Color():
- # 随机颜色,雷达图默认非随机颜色
- rgb = [randint(0, 255), randint(0, 255), randint(0, 255)]
- color = '#'
- for a in rgb:
- color += str(hex(a))[-2:].replace('x',
- '0').upper() # 转换为16进制,upper表示小写(规范化)
- return color
- def is_continuous(data: np.array, f: float = 0.1):
- data = data.tolist()
- l = np.unique(data).tolist()
- try:
- re = len(l) / len(data) >= f or len(data) <= 3
- return re
- except BaseException:
- return False
- def make_Cat(x_data):
- Cat = Categorical_Data()
- for i in range(len(x_data)):
- x1 = x_data[i] # x坐标
- Cat(x1)
- return Cat
- # 根据不同类别绘制x-x分类散点图(可以绘制更多的图)
- def Training_visualization_More_NoCenter(x_trainData, class_, y):
- x_data = x_trainData.transpose
- if len(x_data) == 1:
- x_data = np.array([x_data[0], np.zeros(len(x_data[0]))])
- Cat = make_Cat(x_data)
- o_cList = []
- for i in range(len(x_data)):
- for a in range(len(x_data)):
- if a <= i:
- continue
- x1 = x_data[i] # x坐标
- x1_con = is_continuous(x1)
- x2 = x_data[a] # y坐标
- x2_con = is_continuous(x2)
- o_c = None # 旧的C
- for class_num in range(len(class_)):
- n_class = class_[class_num]
- x_1 = x1[y == n_class].tolist()
- x_2 = x2[y == n_class]
- x_2_new = np.unique(x_2)
- x_2 = x2[y == n_class].tolist()
- # x与散点图不同,这里是纵坐标
- c = (
- Scatter() .add_xaxis(x_2) .add_yaxis(
- f'{n_class}',
- x_1,
- **Label_Set) .set_global_opts(
- title_opts=opts.TitleOpts(
- title=f'[{a}-{i}]训练数据散点图'),
- **global_Set,
- yaxis_opts=opts.AxisOpts(
- type_='value' if x1_con else 'category',
- is_scale=True),
- xaxis_opts=opts.AxisOpts(
- type_='value' if x2_con else 'category',
- is_scale=True)))
- c.add_xaxis(x_2_new)
- if o_c is None:
- o_c = c
- else:
- o_c = o_c.overlap(c)
- o_cList.append(o_c)
- means, x_range, Type = Cat.get()
- return o_cList, means, x_range, Type
- # 根据不同类别绘制x-x分类散点图(可以绘制更多的图)
- def Training_visualization_More(x_trainData, class_, y, center):
- x_data = x_trainData.transpose
- if len(x_data) == 1:
- x_data = np.array([x_data[0], np.zeros(len(x_data[0]))])
- Cat = make_Cat(x_data)
- o_cList = []
- for i in range(len(x_data)):
- for a in range(len(x_data)):
- if a <= i:
- continue
- x1 = x_data[i] # x坐标
- x1_con = is_continuous(x1)
- x2 = x_data[a] # y坐标
- x2_con = is_continuous(x2)
- o_c = None # 旧的C
- for class_num in range(len(class_)):
- n_class = class_[class_num]
- x_1 = x1[y == n_class].tolist()
- x_2 = x2[y == n_class]
- x_2_new = np.unique(x_2)
- x_2 = x2[y == n_class].tolist()
- # x与散点图不同,这里是纵坐标
- c = (
- Scatter() .add_xaxis(x_2) .add_yaxis(
- f'{n_class}',
- x_1,
- **Label_Set) .set_global_opts(
- title_opts=opts.TitleOpts(
- title=f'[{a}-{i}]训练数据散点图'),
- **global_Set,
- yaxis_opts=opts.AxisOpts(
- type_='value' if x1_con else 'category',
- is_scale=True),
- xaxis_opts=opts.AxisOpts(
- type_='value' if x2_con else 'category',
- is_scale=True)))
- c.add_xaxis(x_2_new)
- # 添加簇中心
- try:
- center_x_2 = [center[class_num][a]]
- except BaseException:
- center_x_2 = [0]
- b = (
- Scatter() .add_xaxis(center_x_2) .add_yaxis(
- f'[{n_class}]中心',
- [
- center[class_num][i]],
- **Label_Set,
- symbol='triangle') .set_global_opts(
- title_opts=opts.TitleOpts(
- title='簇中心'),
- **global_Set,
- yaxis_opts=opts.AxisOpts(
- type_='value' if x1_con else 'category',
- is_scale=True),
- xaxis_opts=opts.AxisOpts(
- type_='value' if x2_con else 'category',
- is_scale=True)))
- c.overlap(b)
- if o_c is None:
- o_c = c
- else:
- o_c = o_c.overlap(c)
- o_cList.append(o_c)
- means, x_range, Type = Cat.get()
- return o_cList, means, x_range, Type
- # 根据不同类别绘制x-x分类散点图(可以绘制更多的图)
- def Training_visualization_Center(x_trainData, class_, y, center):
- x_data = x_trainData.transpose
- if len(x_data) == 1:
- x_data = np.array([x_data[0], np.zeros(len(x_data[0]))])
- Cat = make_Cat(x_data)
- o_cList = []
- for i in range(len(x_data)):
- x1 = x_data[i] # x坐标
- x1_con = is_continuous(x1)
- if i == 0:
- continue
- x2 = x_data[i - 1] # y坐标
- x2_con = is_continuous(x2)
- o_c = None # 旧的C
- for class_num in range(len(class_)):
- n_class = class_[class_num]
- x_1 = x1[y == n_class].tolist()
- x_2 = x2[y == n_class]
- x_2_new = np.unique(x_2)
- x_2 = x2[y == n_class].tolist()
- # x与散点图不同,这里是纵坐标
- c = (
- Scatter() .add_xaxis(x_2) .add_yaxis(
- f'{n_class}',
- x_1,
- **Label_Set) .set_global_opts(
- title_opts=opts.TitleOpts(
- title=f'[{i-1}-{i}]训练数据散点图'),
- **global_Set,
- yaxis_opts=opts.AxisOpts(
- type_='value' if x1_con else 'category',
- is_scale=True),
- xaxis_opts=opts.AxisOpts(
- type_='value' if x2_con else 'category',
- is_scale=True)))
- c.add_xaxis(x_2_new)
- # 添加簇中心
- try:
- center_x_2 = [center[class_num][i - 1]]
- except BaseException:
- center_x_2 = [0]
- b = (
- Scatter() .add_xaxis(center_x_2) .add_yaxis(
- f'[{n_class}]中心',
- [
- center[class_num][i]],
- **Label_Set,
- symbol='triangle') .set_global_opts(
- title_opts=opts.TitleOpts(
- title='簇中心'),
- **global_Set,
- yaxis_opts=opts.AxisOpts(
- type_='value' if x1_con else 'category',
- is_scale=True),
- xaxis_opts=opts.AxisOpts(
- type_='value' if x2_con else 'category',
- is_scale=True)))
- c.overlap(b)
- if o_c is None:
- o_c = c
- else:
- o_c = o_c.overlap(c)
- o_cList.append(o_c)
- means, x_range, Type = Cat.get()
- return o_cList, means, x_range, Type
- def Training_visualization(x_trainData, class_, y): # 根据不同类别绘制x-x分类散点图
- x_data = x_trainData.transpose
- if len(x_data) == 1:
- x_data = np.array([x_data[0], np.zeros(len(x_data[0]))])
- Cat = make_Cat(x_data)
- o_cList = []
- for i in range(len(x_data)):
- x1 = x_data[i] # x坐标
- x1_con = is_continuous(x1)
- if i == 0:
- continue
- x2 = x_data[i - 1] # y坐标
- x2_con = is_continuous(x2)
- o_c = None # 旧的C
- for n_class in class_:
- x_1 = x1[y == n_class].tolist()
- x_2 = x2[y == n_class]
- x_2_new = np.unique(x_2)
- x_2 = x2[y == n_class].tolist()
- # x与散点图不同,这里是纵坐标
- c = (
- Scatter() .add_xaxis(x_2) .add_yaxis(
- f'{n_class}',
- x_1,
- **Label_Set) .set_global_opts(
- title_opts=opts.TitleOpts(
- title='训练数据散点图'),
- **global_Set,
- yaxis_opts=opts.AxisOpts(
- type_='value' if x1_con else 'category',
- is_scale=True),
- xaxis_opts=opts.AxisOpts(
- type_='value' if x2_con else 'category',
- is_scale=True)))
- c.add_xaxis(x_2_new)
- if o_c is None:
- o_c = c
- else:
- o_c = o_c.overlap(c)
- o_cList.append(o_c)
- means, x_range, Type = Cat.get()
- return o_cList, means, x_range, Type
- def Training_visualization_NoClass(x_trainData): # 根据绘制x-x分类散点图(无类别)
- x_data = x_trainData.transpose
- if len(x_data) == 1:
- x_data = np.array([x_data[0], np.zeros(len(x_data[0]))])
- Cat = make_Cat(x_data)
- o_cList = []
- for i in range(len(x_data)):
- x1 = x_data[i] # x坐标
- x1_con = is_continuous(x1)
- if i == 0:
- continue
- x2 = x_data[i - 1] # y坐标
- x2_con = is_continuous(x2)
- x2_new = np.unique(x2)
- # x与散点图不同,这里是纵坐标
- c = (
- Scatter() .add_xaxis(x2) .add_yaxis(
- '',
- x1.tolist(),
- **Label_Set) .set_global_opts(
- title_opts=opts.TitleOpts(
- title='训练数据散点图'),
- **global_Leg,
- yaxis_opts=opts.AxisOpts(
- type_='value' if x1_con else 'category',
- is_scale=True),
- xaxis_opts=opts.AxisOpts(
- type_='value' if x2_con else 'category',
- is_scale=True)))
- c.add_xaxis(x2_new)
- o_cList.append(c)
- means, x_range, Type = Cat.get()
- return o_cList, means, x_range, Type
- def Training_W(x_trainData, class_, y, w_list, b_list, means: list): # 针对分类问题绘制决策边界
- x_data = x_trainData.transpose
- if len(x_data) == 1:
- x_data = np.array([x_data[0], np.zeros(len(x_data[0]))])
- o_cList = []
- means.append(0)
- means = np.array(means)
- for i in range(len(x_data)):
- if i == 0:
- continue
- x1_con = is_continuous(x_data[i])
- x2 = x_data[i - 1] # y坐标
- x2_con = is_continuous(x2)
- o_c = None # 旧的C
- for class_num in range(len(class_)):
- n_class = class_[class_num]
- x2_new = np.unique(x2[y == n_class])
- # x与散点图不同,这里是纵坐标
- # 加入这个判断是为了解决sklearn历史遗留问题
- if len(class_) == 2: # 二分类问题
- if class_num == 0:
- continue
- w = w_list[0]
- b = b_list[0]
- else: # 多分类问题
- w = w_list[class_num]
- b = b_list[class_num]
- if x2_con:
- x2_new = np.array(make_list(x2_new.min(), x2_new.max(), 5))
- w = np.append(w, 0)
- y_data = -(x2_new * w[i - 1]) / w[i] + b + (means[:i - 1] * w[:i - 1]).sum() + (
- means[i + 1:] * w[i + 1:]).sum() # 假设除了两个特征意外,其余特征均为means列表的数值
- c = (
- line() .add_xaxis(x2_new) .add_yaxis(
- f"决策边界:{n_class}=>[{i}]",
- y_data.tolist(),
- is_smooth=True,
- **Label_Set) .set_global_opts(
- title_opts=opts.TitleOpts(
- title=f"系数w曲线"),
- **global_Set,
- yaxis_opts=opts.AxisOpts(
- type_='value' if x1_con else 'category',
- is_scale=True),
- xaxis_opts=opts.AxisOpts(
- type_='value' if x2_con else 'category',
- is_scale=True)))
- if o_c is None:
- o_c = c
- else:
- o_c = o_c.overlap(c)
- # 下面不要接任何代码,因为上面会continue
- o_cList.append(o_c)
- return o_cList
- def Regress_W(x_trainData, y, w: np.array, b, means: list): # 针对回归问题(y-x图)
- x_data = x_trainData.transpose
- if len(x_data) == 1:
- x_data = np.array([x_data[0], np.zeros(len(x_data[0]))])
- o_cList = []
- means.append(0) # 确保mean[i+1]不会超出index
- means = np.array(means)
- w = np.append(w, 0)
- for i in range(len(x_data)):
- x1 = x_data[i]
- x1_con = is_continuous(x1)
- if x1_con:
- x1 = np.array(make_list(x1.min(), x1.max(), 5))
- x1_new = np.unique(x1)
- # 假设除了两个特征意外,其余特征均为means列表的数值
- y_data = x1_new * \
- w[i] + b + (means[:i] * w[:i]).sum() + (means[i + 1:] * w[i + 1:]).sum()
- y_con = is_continuous(y_data)
- c = (
- line() .add_xaxis(x1_new) .add_yaxis(
- f"拟合结果=>[{i}]",
- y_data.tolist(),
- is_smooth=True,
- **Label_Set) .set_global_opts(
- title_opts=opts.TitleOpts(
- title=f"系数w曲线"),
- **global_Set,
- yaxis_opts=opts.AxisOpts(
- type_='value' if y_con else None,
- is_scale=True),
- xaxis_opts=opts.AxisOpts(
- type_='value' if x1_con else None,
- is_scale=True)))
- o_cList.append(c)
- return o_cList
- def regress_visualization(x_trainData, y): # y-x数据图
- x_data = x_trainData.transpose
- y_con = is_continuous(y)
- Cat = make_Cat(x_data)
- o_cList = []
- try:
- visualmap_opts = opts.VisualMapOpts(
- is_show=True,
- max_=int(
- y.max()) + 1,
- min_=int(
- y.min()),
- pos_right='3%')
- except BaseException:
- visualmap_opts = None
- y_con = False
- for i in range(len(x_data)):
- x1 = x_data[i] # x坐标
- x1_con = is_continuous(x1)
- # 不转换成list因为保持dtype的精度,否则绘图会出现各种问题(数值重复)
- if not y_con and x1_con: # y不是连续的但x1连续,ry和ry_con是保护y的
- ry_con, x1_con = x1_con, y_con
- x1, ry = y, x1
- else:
- ry_con = y_con
- ry = y
- c = (
- Scatter()
- .add_xaxis(x1.tolist()) # 研究表明,这个是横轴
- .add_yaxis('数据', ry.tolist(), **Label_Set)
- .set_global_opts(title_opts=opts.TitleOpts(title="预测类型图"), **global_Set,
- yaxis_opts=opts.AxisOpts(
- type_='value' if ry_con else 'category', is_scale=True),
- xaxis_opts=opts.AxisOpts(
- type_='value' if x1_con else 'category', is_scale=True),
- visualmap_opts=visualmap_opts
- )
- )
- c.add_xaxis(np.unique(x1))
- o_cList.append(c)
- means, x_range, Type = Cat.get()
- return o_cList, means, x_range, Type
- def Feature_visualization(x_trainData, data_name=''): # x-x数据图
- seeting = global_Set if data_name else global_Leg
- x_data = x_trainData.transpose
- only = False
- if len(x_data) == 1:
- x_data = np.array([x_data[0], np.zeros(len(x_data[0]))])
- only = True
- o_cList = []
- for i in range(len(x_data)):
- for a in range(len(x_data)):
- if a <= i:
- continue # 重复内容,跳过
- x1 = x_data[i] # x坐标
- x1_con = is_continuous(x1)
- x2 = x_data[a] # y坐标
- x2_con = is_continuous(x2)
- x2_new = np.unique(x2)
- if only:
- x2_con = False
- # x与散点图不同,这里是纵坐标
- c = (
- Scatter() .add_xaxis(x2) .add_yaxis(
- data_name,
- x1,
- **Label_Set) .set_global_opts(
- title_opts=opts.TitleOpts(
- title=f'[{i}-{a}]数据散点图'),
- **seeting,
- yaxis_opts=opts.AxisOpts(
- type_='value' if x1_con else 'category',
- is_scale=True),
- xaxis_opts=opts.AxisOpts(
- type_='value' if x2_con else 'category',
- is_scale=True)))
- c.add_xaxis(x2_new)
- o_cList.append(c)
- return o_cList
- def Feature_visualization_Format(x_trainData, data_name=''): # x-x数据图
- seeting = global_Set if data_name else global_Leg
- x_data = x_trainData.transpose
- only = False
- if len(x_data) == 1:
- x_data = np.array([x_data[0], np.zeros(len(x_data[0]))])
- only = True
- o_cList = []
- for i in range(len(x_data)):
- for a in range(len(x_data)):
- if a <= i:
- continue # 重复内容,跳过(a读取的是i后面的)
- x1 = x_data[i] # x坐标
- x1_con = is_continuous(x1)
- x2 = x_data[a] # y坐标
- x2_con = is_continuous(x2)
- x2_new = np.unique(x2)
- x1_list = x1.astype(np.str).tolist()
- for i in range(len(x1_list)):
- x1_list[i] = [x1_list[i], f'特征{i}']
- if only:
- x2_con = False
- # x与散点图不同,这里是纵坐标
- c = (
- Scatter() .add_xaxis(x2) .add_yaxis(
- data_name,
- x1_list,
- **Label_Set) .set_global_opts(
- title_opts=opts.TitleOpts(
- title=f'[{i}-{a}]数据散点图'),
- **seeting,
- yaxis_opts=opts.AxisOpts(
- type_='value' if x1_con else 'category',
- is_scale=True),
- xaxis_opts=opts.AxisOpts(
- type_='value' if x2_con else 'category',
- is_scale=True),
- tooltip_opts=opts.TooltipOpts(
- is_show=True,
- axis_pointer_type="cross",
- formatter="{c}")))
- c.add_xaxis(x2_new)
- o_cList.append(c)
- return o_cList
- def Discrete_Feature_visualization(x_trainData, data_name=''): # 必定离散x-x数据图
- seeting = global_Set if data_name else global_Leg
- x_data = x_trainData.transpose
- if len(x_data) == 1:
- x_data = np.array([x_data[0], np.zeros(len(x_data[0]))])
- o_cList = []
- for i in range(len(x_data)):
- for a in range(len(x_data)):
- if a <= i:
- continue # 重复内容,跳过
- x1 = x_data[i] # x坐标
- x2 = x_data[a] # y坐标
- x2_new = np.unique(x2)
- # x与散点图不同,这里是纵坐标
- c = (
- Scatter() .add_xaxis(x2) .add_yaxis(
- data_name,
- x1,
- **Label_Set) .set_global_opts(
- title_opts=opts.TitleOpts(
- title=f'[{i}-{a}]数据散点图'),
- **seeting,
- yaxis_opts=opts.AxisOpts(
- type_='category',
- is_scale=True),
- xaxis_opts=opts.AxisOpts(
- type_='category',
- is_scale=True)))
- c.add_xaxis(x2_new)
- o_cList.append(c)
- return o_cList
- def Conversion_control(y_data, x_data, tab): # 合并两x-x图
- if isinstance(x_data, np.ndarray) and isinstance(y_data, np.ndarray):
- get_x = Feature_visualization(x_data, '原数据') # 原来
- get_y = Feature_visualization(y_data, '转换数据') # 转换
- for i in range(len(get_x)):
- tab.add(get_x[i].overlap(get_y[i]), f'[{i}]数据x-x散点图')
- return tab
- def Conversion_Separate(y_data, x_data, tab): # 并列显示两x-x图
- if isinstance(x_data, np.ndarray) and isinstance(y_data, np.ndarray):
- get_x = Feature_visualization(x_data, '原数据') # 原来
- get_y = Feature_visualization(y_data, '转换数据') # 转换
- for i in range(len(get_x)):
- try:
- tab.add(get_x[i], f'[{i}]数据x-x散点图')
- except IndexError:
- pass
- try:
- tab.add(get_y[i], f'[{i}]变维数据x-x散点图')
- except IndexError:
- pass
- return tab
- def Conversion_Separate_Format(y_data, tab): # 并列显示两x-x图
- if isinstance(y_data, np.ndarray):
- get_y = Feature_visualization_Format(y_data, '转换数据') # 转换
- for i in range(len(get_y)):
- tab.add(get_y[i], f'[{i}]变维数据x-x散点图')
- return tab
- def Conversion_SeparateWH(w_data, h_data, tab): # 并列显示两x-x图
- if isinstance(w_data, np.ndarray) and isinstance(w_data, np.ndarray):
- get_x = Feature_visualization_Format(w_data, 'W矩阵数据') # 原来
- get_y = Feature_visualization(
- h_data.transpose, 'H矩阵数据') # 转换(先转T,再转T变回原样,W*H是横对列)
- print(h_data)
- print(w_data)
- print(h_data.transpose)
- for i in range(len(get_x)):
- try:
- tab.add(get_x[i], f'[{i}]W矩阵x-x散点图')
- except IndexError:
- pass
- try:
- tab.add(get_y[i], f'[{i}]H.T矩阵x-x散点图')
- except IndexError:
- pass
- return tab
- def make_bar(name, value, tab): # 绘制柱状图
- c = (
- Bar()
- .add_xaxis([f'[{i}]特征' for i in range(len(value))])
- .add_yaxis(name, value, **Label_Set)
- .set_global_opts(title_opts=opts.TitleOpts(title='系数w柱状图'), **global_Set)
- )
- tab.add(c, name)
- def judging_Digits(num: (int, float)): # 查看小数位数
- a = str(abs(num)).split('.')[0]
- if a == '':
- raise ValueError
- return len(a)
- class Learner:
- def __init__(self, *args, **kwargs):
- self.numpy_Dic = {} # name:numpy
- self.Fucn_Add() # 制作Func_Dic
- def Add_Form(self, data: np.array, name):
- name = f'{name}[{len(self.numpy_Dic)}]'
- self.numpy_Dic[name] = data
- def read_csv(self, Dic, name, encoding='utf-8', str_must=False, sep=','):
- type_ = np.str if str_must else np.float
- pf_data = read_csv(Dic, encoding=encoding, delimiter=sep, header=None)
- try:
- data = pf_data.to_numpy(dtype=type_)
- except ValueError:
- data = pf_data.to_numpy(dtype=np.str)
- if data.ndim == 1:
- data = np.expand_dims(data, axis=1)
- self.Add_Form(data, name)
- return data
- def Add_Python(self, Text, sheet_name):
- name = {}
- name.update(globals().copy())
- name.update(locals().copy())
- exec(Text, name)
- exec('get = Creat()', name)
- if isinstance(name['get'], np.array): # 已经是DataFram
- get = name['get']
- else:
- try:
- get = np.array(name['get'])
- except BaseException:
- get = np.array([name['get']])
- self.Add_Form(get, sheet_name)
- return get
- def get_Form(self) -> dict:
- return self.numpy_Dic.copy()
- def get_Sheet(self, name) -> np.array:
- return self.numpy_Dic[name].copy()
- def to_CSV(self, Dic: str, name, sep) -> str:
- get = self.get_Sheet(name)
- np.savetxt(Dic, get, delimiter=sep)
- return Dic
- def to_Html_One(self, name, Dic=''):
- if Dic == '':
- Dic = f'{name}.html'
- get = self.get_Sheet(name)
- if get.ndim == 1:
- get = np.expand_dims(get, axis=1)
- get = get.tolist()
- for i in range(len(get)):
- get[i] = [i + 1] + get[i]
- headers = [i for i in range(len(get[0]))]
- table = Table_Fisrt()
- table.add(
- headers,
- get).set_global_opts(
- title_opts=opts.ComponentTitleOpts(
- title=f"表格:{name}",
- subtitle="CoTan~机器学习:查看数据"))
- table.render(Dic)
- return Dic
- def to_Html(self, name, Dic='', type_=0):
- if Dic == '':
- Dic = f'{name}.html'
- # 把要画的sheet放到第一个
- Sheet_Dic = self.get_Form()
- del Sheet_Dic[name]
- Sheet_list = [name] + list(Sheet_Dic.keys())
- class TAB_F:
- def __init__(self, q):
- self.tab = q # 一个Tab
- def render(self, Dic):
- return self.tab.render(Dic)
- # 生成一个显示页面
- if type_ == 0:
- class TAB(TAB_F):
- def add(self, table, k, *f):
- self.tab.add(table, k)
- tab = TAB(tab_First(page_title='CoTan:查看表格')) # 一个Tab
- elif type_ == 1:
- class TAB(TAB_F):
- def add(self, table, *k):
- self.tab.add(table)
- tab = TAB(
- Page(
- page_title='CoTan:查看表格',
- layout=Page.DraggablePageLayout))
- else:
- class TAB(TAB_F):
- def add(self, table, *k):
- self.tab.add(table)
- tab = TAB(
- Page(
- page_title='CoTan:查看表格',
- layout=Page.SimplePageLayout))
- # 迭代添加内容
- for name in Sheet_list:
- get = self.get_Sheet(name)
- if get.ndim == 1:
- get = np.expand_dims(get, axis=1)
- get = get.tolist()
- for i in range(len(get)):
- get[i] = [i + 1] + get[i]
- headers = [i for i in range(len(get[0]))]
- table = Table_Fisrt()
- table.add(
- headers,
- get).set_global_opts(
- title_opts=opts.ComponentTitleOpts(
- title=f"表格:{name}",
- subtitle="CoTan~机器学习:查看数据"))
- tab.add(table, f'表格:{name}')
- tab.render(Dic)
- return Dic
- def Merge(self, name, axis=0): # aiis:0-横向合并(hstack),1-纵向合并(vstack),2-深度合并
- sheet_list = []
- for i in name:
- sheet_list.append(self.get_Sheet(i))
- get = {0: np.hstack, 1: np.vstack, 2: np.dstack}[axis](sheet_list)
- self.Add_Form(np.array(get), f'{name[0]}合成')
- def Split(self, name, split=2, axis=0): # aiis:0-横向分割(hsplit),1-纵向分割(vsplit)
- sheet = self.get_Sheet(name)
- get = {0: np.hsplit, 1: np.vsplit, 2: np.dsplit}[axis](sheet, split)
- for i in get:
- self.Add_Form(i, f'{name[0]}分割')
- def Two_Split(self, name, split, axis): # 二分切割(0-横向,1-纵向)
- sheet = self.get_Sheet(name)
- try:
- split = float(eval(split))
- if split < 1:
- split = int(split * len(sheet) if axis == 1 else len(sheet[0]))
- else:
- raise Exception
- except BaseException:
- split = int(split)
- if axis == 0:
- self.Add_Form(sheet[:, split:], f'{name[0]}分割')
- self.Add_Form(sheet[:, :split], f'{name[0]}分割')
- def Deep(self, sheet: np.ndarray):
- return sheet.ravel()
- def Down_Ndim(self, sheet: np.ndarray): # 横向
- down_list = []
- for i in sheet:
- down_list.append(i.ravel())
- return np.array(down_list)
- def LongitudinalDown_Ndim(self, sheet: np.ndarray): # 纵向
- down_list = []
- for i in range(len(sheet[0])):
- down_list.append(sheet[:, i].ravel())
- return np.array(down_list).T
- def Reval(self, name, axis): # axis:0-横向,1-纵向(带.T),2-深度
- sheet = self.get_Sheet(name)
- self.Add_Form({0: self.Down_Ndim, 1: self.LongitudinalDown_Ndim, 2: self.Deep}[
- axis](sheet).copy(), f'{name}伸展')
- def Del_Ndim(self, name): # 删除无用维度
- sheet = self.get_Sheet(name)
- self.Add_Form(np.squeeze(sheet), f'{name}降维')
- def T(self, name, Func: list):
- sheet = self.get_Sheet(name)
- if sheet.ndim <= 2:
- self.Add_Form(sheet.transpose.copy(), f'{name}.T')
- else:
- self.Add_Form(np.transpose(sheet, Func).copy(), f'{name}.T')
- def reShape(self, name, shape: list):
- sheet = self.get_Sheet(name)
- self.Add_Form(sheet.reshape(shape).copy(), f'{name}.r')
- def Fucn_Add(self):
- self.Func_Dic = {
- 'abs': lambda x, y: np.abs(x),
- 'sqrt': lambda x, y: np.sqrt(x),
- 'pow': lambda x, y: x**y,
- 'loge': lambda x, y: np.log(x),
- 'log10': lambda x, y: np.log10(x),
- 'ceil': lambda x, y: np.ceil(x),
- 'floor': lambda x, y: np.floor(x),
- 'rint': lambda x, y: np.rint(x),
- 'sin': lambda x, y: np.sin(x),
- 'cos': lambda x, y: np.cos(x),
- 'tan': lambda x, y: np.tan(x),
- 'tanh': lambda x, y: np.tanh(x),
- 'sinh': lambda x, y: np.sinh(x),
- 'cosh': lambda x, y: np.cosh(x),
- 'asin': lambda x, y: np.arcsin(x),
- 'acos': lambda x, y: np.arccos(x),
- 'atan': lambda x, y: np.arctan(x),
- 'atanh': lambda x, y: np.arctanh(x),
- 'asinh': lambda x, y: np.arcsinh(x),
- 'acosh': lambda x, y: np.arccosh(x),
- 'add': lambda x, y: x + y, # 矩阵或元素
- 'sub': lambda x, y: x - y, # 矩阵或元素
- 'mul': lambda x, y: np.multiply(x, y), # 元素级别
- 'matmul': lambda x, y: np.matmul(x, y), # 矩阵
- 'dot': lambda x, y: np.dot(x, y), # 矩阵
- 'div': lambda x, y: x / y,
- 'div_floor': lambda x, y: np.floor_divide(x, y),
- 'power': lambda x, y: np.power(x, y), # 元素级
- }
- def Cul_Numpy(self, data, data_type, Func):
- if 1 not in data_type:
- raise Exception
- func = self.Func_Dic.get(Func, lambda x, y: x)
- args_data = []
- for i in range(len(data)):
- if data_type[i] == 0:
- args_data.append(data[i])
- else:
- args_data.append(self.get_Sheet(data[i]))
- get = func(*args_data)
- self.Add_Form(get, f'{Func}({data[0]},{data[1]})')
- return get
- class Study_MachineBase:
- def __init__(self, *args, **kwargs):
- self.Model = None
- self.have_Fit = False
- self.have_Predict = False
- self.x_trainData = None
- self.y_trainData = None
- # 有监督学习专有的testData
- self.x_testData = None
- self.y_testData = None
- # 记录这两个是为了克隆
- def Fit(self, x_data, y_data, split=0.3, Increment=True, **kwargs):
- y_data = y_data.ravel()
- try:
- if self.x_trainData is None or not Increment:
- raise Exception
- self.x_trainData = np.vstack(x_data, self.x_trainData)
- self.y_trainData = np.vstack(y_data, self.y_trainData)
- except BaseException:
- self.x_trainData = x_data.copy()
- self.y_trainData = y_data.copy()
- x_train, x_test, y_train, y_test = train_test_split(
- x_data, y_data, test_size=split)
- try: # 增量式训练
- if not Increment:
- raise Exception
- self.Model.partial_fit(x_data, y_data)
- except BaseException:
- self.Model.fit(self.x_trainData, self.y_trainData)
- train_score = self.Model.score(x_train, y_train)
- test_score = self.Model.score(x_test, y_test)
- self.have_Fit = True
- return train_score, test_score
- def Score(self, x_data, y_data):
- Score = self.Model.score(x_data, y_data)
- return Score
- def Class_Score(self, Dic, x_data: np.ndarray, y_Really: np.ndarray):
- y_Really = y_Really.ravel()
- y_Predict = self.Predict(x_data)[0]
- Accuracy = self._Accuracy(y_Predict, y_Really)
- Recall, class_ = self._Macro(y_Predict, y_Really)
- Precision, class_ = self._Macro(y_Predict, y_Really, 1)
- F1, class_ = self._Macro(y_Predict, y_Really, 2)
- Confusion_matrix, class_ = self._Confusion_matrix(y_Predict, y_Really)
- kappa = self._Kappa_score(y_Predict, y_Really)
- tab = Tab()
- def gauge_base(name: str, value: float) -> Gauge:
- c = (
- Gauge()
- .add("", [(name, round(value * 100, 2))], min_=0, max_=100)
- .set_global_opts(title_opts=opts.TitleOpts(title=name))
- )
- return c
- tab.add(gauge_base('准确率', Accuracy), '准确率')
- tab.add(gauge_base('kappa', kappa), 'kappa')
- def Bar_base(name, value) -> Bar:
- c = (
- Bar() .add_xaxis(class_) .add_yaxis(
- name, value, **Label_Set) .set_global_opts(
- title_opts=opts.TitleOpts(
- title=name), **global_Set))
- return c
- tab.add(Bar_base('精确率', Precision.tolist()), '精确率')
- tab.add(Bar_base('召回率', Recall.tolist()), '召回率')
- tab.add(Bar_base('F1', F1.tolist()), 'F1')
- def heatmap_base(name, value, max_, min_, show) -> HeatMap:
- c = (
- HeatMap() .add_xaxis(class_) .add_yaxis(
- name,
- class_,
- value,
- label_opts=opts.LabelOpts(
- is_show=show,
- position='inside')) .set_global_opts(
- title_opts=opts.TitleOpts(
- title=name),
- **global_Set,
- visualmap_opts=opts.VisualMapOpts(
- max_=max_,
- min_=min_,
- pos_right='3%')))
- return c
- value = [[class_[i], class_[j], float(Confusion_matrix[i, j])] for i in range(
- len(class_)) for j in range(len(class_))]
- tab.add(
- heatmap_base(
- '混淆矩阵', value, float(
- Confusion_matrix.max()), float(
- Confusion_matrix.min()), len(class_) < 7), '混淆矩阵')
- desTo_CSV(Dic, '混淆矩阵', Confusion_matrix, class_, class_)
- desTo_CSV(
- Dic, '评分', [
- Precision, Recall, F1], class_, [
- '精确率', '召回率', 'F1'])
- save = Dic + r'/分类模型评估.HTML'
- tab.render(save)
- return save,
- def _Accuracy(self, y_Predict, y_Really): # 准确率
- return accuracy_score(y_Really, y_Predict)
- def _Macro(self, y_Predict, y_Really, func=0):
- Func = [recall_score, precision_score, f1_score] # 召回率,精确率和f1
- class_ = np.unique(y_Really).tolist()
- result = (Func[func](y_Really, y_Predict, class_, average=None))
- return result, class_
- def _Confusion_matrix(self, y_Predict, y_Really): # 混淆矩阵
- class_ = np.unique(y_Really).tolist()
- return confusion_matrix(y_Really, y_Predict), class_
- def _Kappa_score(self, y_Predict, y_Really):
- return cohen_kappa_score(y_Really, y_Predict)
- def Regression_Score(self, Dic, x_data: np.ndarray, y_Really: np.ndarray):
- y_Really = y_Really.ravel()
- y_Predict = self.Predict(x_data)[0]
- tab = Tab()
- MSE = self._MSE(y_Predict, y_Really)
- MAE = self._MAE(y_Predict, y_Really)
- r2_Score = self._R2_Score(y_Predict, y_Really)
- RMSE = self._RMSE(y_Predict, y_Really)
- tab.add(make_Tab(['MSE', 'MAE', 'RMSE', 'r2_Score'],
- [[MSE, MAE, RMSE, r2_Score]]), '评估数据')
- save = Dic + r'/回归模型评估.HTML'
- tab.render(save)
- return save,
- def Clusters_Score(self, Dic, x_data: np.ndarray, *args):
- y_Predict = self.Predict(x_data)[0]
- tab = Tab()
- Coefficient, Coefficient_array = self._Coefficient_clustering(
- x_data, y_Predict)
- def gauge_base(name: str, value: float) -> Gauge:
- c = (Gauge() .add("", [(name, round(value * 100, 2))], min_=0, max_=10**(
- judging_Digits(value * 100))) .set_global_opts(title_opts=opts.TitleOpts(title=name)))
- return c
- def Bar_base(name, value, xaxis) -> Bar:
- c = (
- Bar() .add_xaxis(xaxis) .add_yaxis(
- name, value, **Label_Set) .set_global_opts(
- title_opts=opts.TitleOpts(
- title=name), **global_Set))
- return c
- tab.add(gauge_base('平均轮廓系数', Coefficient), '平均轮廓系数')
- def Bar_(Coefficient_array, name='数据轮廓系数'):
- xaxis = [f'数据{i}' for i in range(len(Coefficient_array))]
- value = Coefficient_array.tolist()
- tab.add(Bar_base(name, value, xaxis), name)
- n = 20
- if len(Coefficient_array) <= n:
- Bar_(Coefficient_array)
- elif len(Coefficient_array) <= n**2:
- a = 0
- while a <= len(Coefficient_array):
- b = a + n
- if b >= len(Coefficient_array):
- b = len(Coefficient_array) + 1
- Cofe_array = Coefficient_array[a:b]
- Bar_(Cofe_array, f'{a}-{b}数据轮廓系数')
- a += n
- else:
- split = np.hsplit(Coefficient_array, n)
- a = 0
- for Cofe_array in split:
- Bar_(Cofe_array, f'{a}%-{a + n}%数据轮廓系数')
- a += n
- save = Dic + r'/聚类模型评估.HTML'
- tab.render(save)
- return save,
- def _MSE(self, y_Predict, y_Really): # 均方误差
- return mean_squared_error(y_Really, y_Predict)
- def _MAE(self, y_Predict, y_Really): # 中值绝对误差
- return median_absolute_error(y_Really, y_Predict)
- def _R2_Score(self, y_Predict, y_Really): # 中值绝对误差
- return r2_score(y_Really, y_Predict)
- def _RMSE(self, y_Predict, y_Really): # 中值绝对误差
- return self._MSE(y_Predict, y_Really) ** 0.5
- def _Coefficient_clustering(self, x_data, y_Predict):
- means_score = silhouette_score(x_data, y_Predict)
- outline_score = silhouette_samples(x_data, y_Predict)
- return means_score, outline_score
- def Predict(self, x_data, *args, **kwargs):
- self.x_testData = x_data.copy()
- y_Predict = self.Model.predict(x_data)
- self.y_testData = y_Predict.copy()
- self.have_Predict = True
- return y_Predict, '预测'
- def Des(self, Dic, *args, **kwargs):
- return (Dic,)
- class prep_Base(Study_MachineBase): # 不允许第二次训练
- def __init__(self, *args, **kwargs):
- super(prep_Base, self).__init__(*args, **kwargs)
- self.Model = None
- def Fit(self, x_data, y_data, Increment=True, *args, **kwargs):
- if not self.have_Predict: # 不允许第二次训练
- y_data = y_data.ravel()
- try:
- if self.x_trainData is None or not Increment:
- raise Exception
- self.x_trainData = np.vstack(x_data, self.x_trainData)
- self.y_trainData = np.vstack(y_data, self.y_trainData)
- except BaseException:
- self.x_trainData = x_data.copy()
- self.y_trainData = y_data.copy()
- try: # 增量式训练
- if not Increment:
- raise Exception
- self.Model.partial_fit(x_data, y_data)
- except BaseException:
- self.Model.fit(self.x_trainData, self.y_trainData)
- self.have_Fit = True
- return 'None', 'None'
- def Predict(self, x_data, *args, **kwargs):
- self.x_testData = x_data.copy()
- x_Predict = self.Model.transform(x_data)
- self.y_testData = x_Predict.copy()
- self.have_Predict = True
- return x_Predict, '特征工程'
- def Score(self, x_data, y_data):
- return 'None' # 没有score
- class Unsupervised(prep_Base): # 无监督,不允许第二次训练
- def Fit(self, x_data, Increment=True, *args, **kwargs):
- if not self.have_Predict: # 不允许第二次训练
- self.y_trainData = None
- try:
- if self.x_trainData is None or not Increment:
- raise Exception
- self.x_trainData = np.vstack(x_data, self.x_trainData)
- except BaseException:
- self.x_trainData = x_data.copy()
- try: # 增量式训练
- if not Increment:
- raise Exception
- self.Model.partial_fit(x_data)
- except BaseException:
- self.Model.fit(self.x_trainData, self.y_trainData)
- self.have_Fit = True
- return 'None', 'None'
- class UnsupervisedModel(prep_Base): # 无监督
- def Fit(self, x_data, Increment=True, *args, **kwargs):
- self.y_trainData = None
- try:
- if self.x_trainData is None or not Increment:
- raise Exception
- self.x_trainData = np.vstack(x_data, self.x_trainData)
- except BaseException:
- self.x_trainData = x_data.copy()
- try: # 增量式训练
- if not Increment:
- raise Exception
- self.Model.partial_fit(x_data)
- except BaseException:
- self.Model.fit(self.x_trainData, self.y_trainData)
- self.have_Fit = True
- return 'None', 'None'
- class To_PyeBase(Study_MachineBase):
- def __init__(self, args_use, model, *args, **kwargs):
- super(To_PyeBase, self).__init__(*args, **kwargs)
- self.Model = None
- # 记录这两个是为了克隆
- self.k = {}
- self.Model_Name = model
- def Fit(self, x_data, y_data, *args, **kwargs):
- self.x_trainData = x_data.copy()
- self.y_trainData = y_data.ravel().copy()
- self.have_Fit = True
- return 'None', 'None'
- def Predict(self, x_data, *args, **kwargs):
- self.have_Predict = True
- return np.array([]), '请使用训练'
- def Score(self, x_data, y_data):
- return 'None' # 没有score
- def num_str(num, f):
- num = str(round(float(num), f))
- if len(num.replace('.', '')) == f:
- return num
- n = num.split('.')
- if len(n) == 0: # 无小数
- return num + '.' + '0' * (f - len(num))
- else:
- return num + '0' * (f - len(num) + 1) # len(num)多算了一位小数点
- def desTo_CSV(Dic, name, data, columns=None, row=None):
- Dic = Dic + '/' + name + '.csv'
- DataFrame(
- data,
- columns=columns,
- index=row).to_csv(
- Dic,
- header=False if columns is None else True,
- index=False if row is None else True)
- return data
- class Des(To_PyeBase): # 数据分析
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- data = self.x_trainData
- def Cumulative_calculation(data, func, name, tab):
- sum_list = []
- for i in range(len(data)): # 按行迭代数据
- sum_list.append([])
- for a in range(len(data[i])):
- s = num_str(func(data[:i + 1, a]), 8)
- sum_list[-1].append(s)
- desTo_CSV(Dic, f'{name}', sum_list)
- tab.add(make_Tab([f'[{i}]' for i in range(
- len(sum_list[0]))], sum_list), f'{name}')
- def Geometric_mean(x): return np.power(np.prod(x), 1 / len(x)) # 几何平均数
- def Square_mean(x): return np.sqrt(
- np.sum(np.power(x, 2)) / len(x)) # 平方平均数
- def Harmonic_mean(x): return len(x) / np.sum(np.power(x, -1)) # 调和平均数
- Cumulative_calculation(data, np.sum, '累计求和', tab)
- Cumulative_calculation(data, np.var, '累计方差', tab)
- Cumulative_calculation(data, np.std, '累计标准差', tab)
- Cumulative_calculation(data, np.mean, '累计算术平均值', tab)
- Cumulative_calculation(data, Geometric_mean, '累计几何平均值', tab)
- Cumulative_calculation(data, Square_mean, '累计平方平均值', tab)
- Cumulative_calculation(data, Harmonic_mean, '累计调和平均值', tab)
- Cumulative_calculation(data, np.median, '累计中位数', tab)
- Cumulative_calculation(data, np.max, '累计最大值', tab)
- Cumulative_calculation(data, np.min, '累计最小值', tab)
- save = Dic + r'/数据分析.HTML'
- tab.render(save) # 生成HTML
- return save,
- class CORR(To_PyeBase): # 相关性和协方差
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- data = DataFrame(self.x_trainData)
- corr = data.corr().to_numpy() # 相关性
- cov = data.cov().to_numpy() # 协方差
- def HeatMAP(data, name: str, max_, min_):
- x = [f'特征[{i}]' for i in range(len(data))]
- y = [f'特征[{i}]' for i in range(len(data[0]))]
- value = [(f'特征[{i}]', f'特征[{j}]', float(data[i][j]))
- for i in range(len(data)) for j in range(len(data[i]))]
- c = (HeatMap()
- .add_xaxis(x)
- # 如果特征太多则不显示标签
- .add_yaxis(f'数据', y, value, label_opts=opts.LabelOpts(is_show=True if len(x) <= 10 else False, position='inside'))
- .set_global_opts(title_opts=opts.TitleOpts(title='矩阵热力图'), **global_Leg,
- yaxis_opts=opts.AxisOpts(
- is_scale=True, type_='category'), # 'category'
- xaxis_opts=opts.AxisOpts(is_scale=True, type_='category'),
- visualmap_opts=opts.VisualMapOpts(is_show=True, max_=max_, min_=min_, pos_right='3%')) # 显示
- )
- tab.add(c, name)
- HeatMAP(corr, '相关性热力图', 1, -1)
- HeatMAP(cov, '协方差热力图', float(cov.max()), float(cov.min()))
- desTo_CSV(Dic, f'相关性矩阵', corr)
- desTo_CSV(Dic, f'协方差矩阵', cov)
- save = Dic + r'/数据相关性.HTML'
- tab.render(save) # 生成HTML
- return save,
- class View_data(To_PyeBase): # 绘制预测型热力图
- def __init__(self, args_use, Learner, *args, **kwargs): # model表示当前选用的模型类型,Alpha针对正则化的参数
- super(View_data, self).__init__(args_use, Learner, *args, **kwargs)
- self.Model = Learner.Model
- self.Select_Model = None
- self.have_Fit = Learner.have_Fit
- self.Model_Name = 'Select_Model'
- self.Learner = Learner
- self.Learner_name = Learner.Model_Name
- def Fit(self, *args, **kwargs):
- self.have_Fit = True
- return 'None', 'None'
- def Predict(self, x_data, Add_Func=None, *args, **kwargs):
- x_trainData = self.Learner.x_trainData
- y_trainData = self.Learner.y_trainData
- x_name = self.Learner_name
- if x_trainData is not None:
- Add_Func(x_trainData, f'{x_name}:x训练数据')
- try:
- x_testData = self.x_testData
- if x_testData is not None:
- Add_Func(x_testData, f'{x_name}:x测试数据')
- except BaseException:
- pass
- try:
- y_testData = self.y_testData.copy()
- if y_testData is not None:
- Add_Func(y_testData, f'{x_name}:y测试数据')
- except BaseException:
- pass
- self.have_Fit = True
- if y_trainData is None:
- return np.array([]), 'y训练数据'
- return y_trainData, 'y训练数据'
- def Des(self, Dic, *args, **kwargs):
- return Dic,
- class MatrixScatter(To_PyeBase): # 矩阵散点图
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- data = self.x_trainData
- if data.ndim <= 2: # 维度为2
- c = (Scatter()
- .add_xaxis([f'{i}' for i in range(data.shape[1])])
- .set_global_opts(title_opts=opts.TitleOpts(title=f'矩阵散点图'), **global_Leg)
- )
- if data.ndim == 2:
- for num in range(len(data)):
- i = data[num]
- c.add_yaxis(f'{num}', [[f'{num}', x]
- for x in i], color='#FFFFFF')
- else:
- c.add_yaxis(f'0', [[0, x]for x in data], color='#FFFFFF')
- c.set_series_opts(
- label_opts=opts.LabelOpts(
- is_show=True,
- color='#000000',
- position='inside',
- formatter=JsCode("function(params){return params.data[2];}"),
- ))
- elif data.ndim == 3:
- c = (
- Scatter3D() .set_global_opts(
- title_opts=opts.TitleOpts(
- title=f'矩阵散点图'),
- **global_Leg))
- for num in range(len(data)):
- i = data[num]
- for s_num in range(len(i)):
- s = i[s_num]
- y_data = [[num, s_num, x, float(s[x])]
- for x in range(len(s))]
- c.add(
- f'{num}',
- y_data,
- zaxis3d_opts=opts.Axis3DOpts(
- type_="category"))
- c.set_series_opts(
- label_opts=opts.LabelOpts(
- is_show=True,
- color='#000000',
- position='inside',
- formatter=JsCode("function(params){return params.data[3];}")))
- else:
- c = Scatter()
- tab.add(c, '矩阵散点图')
- save = Dic + r'/矩阵散点图.HTML'
- tab.render(save) # 生成HTML
- return save,
- class Cluster_Tree(To_PyeBase): # 聚类树状图
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- x_data = self.x_trainData
- linkage_array = ward(x_data) # self.y_trainData是结果
- dendrogram(linkage_array)
- plt.savefig(Dic + r'/Cluster_graph.png')
- image = Image()
- image.add(
- src=Dic + r'/Cluster_graph.png',
- ).set_global_opts(
- title_opts=opts.ComponentTitleOpts(
- title="聚类树状图"))
- tab.add(image, '聚类树状图')
- save = Dic + r'/聚类树状图.HTML'
- tab.render(save) # 生成HTML
- return save,
- class Class_To_Bar(To_PyeBase): # 类型柱状图
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- x_data = self.x_trainData.transpose
- y_data = self.y_trainData
- class_ = np.unique(y_data).tolist() # 类型
- class_list = []
- for n_class in class_: # 生成class_list(class是1,,也就是二维的,下面会压缩成一维)
- class_list.append(y_data == n_class)
- for num_i in range(len(x_data)): # 迭代每一个特征
- i = x_data[num_i]
- i_con = is_continuous(i)
- if i_con and len(i) >= 11:
- # 存放绘图数据,每一层列表是一个类(leg),第二层是每个x_data
- c_list = [[0] * 10 for _ in class_list]
- start = i.min()
- end = i.max()
- n = (end - start) / 10 # 生成10条柱子
- x_axis = [] # x轴
- num_startEND = 0 # 迭代到第n个
- while num_startEND <= 9: # 把每个特征分为10类进行迭代
- # x_axis添加数据
- x_axis.append(
- f'({num_startEND})[{round(start, 2)}-{round((start + n) if (start + n) <= end or not num_startEND == 9 else end, 2)}]')
- try:
- if num_startEND == 9:
- raise Exception # 执行到第10次时,直接获取剩下的所有
- s = (start <= i) == (i < end) # 布尔索引
- except BaseException: # 因为start + n有超出end的风险
- s = (start <= i) == (i <= end) # 布尔索引
- # n_data = i[s] # 取得现在的特征数据
- for num in range(len(class_list)): # 根据类别进行迭代
- # 取得布尔数组:y_data == n_class也就是输出值为指定类型的bool矩阵,用于切片
- now_class = class_list[num]
- # 切片成和n_data一样的位置一样的形状(now_class就是一个bool矩阵)
- bool_class = now_class[s].ravel()
- # 用len计数 c_list = [[class1的数据],[class2的数据],[]]
- c_list[num][num_startEND] = (int(np.sum(bool_class)))
- num_startEND += 1
- start += n
- else:
- iter_np = np.unique(i)
- # 存放绘图数据,每一层列表是一个类(leg),第二层是每个x_data
- c_list = [[0] * len(iter_np) for _ in class_list]
- x_axis = [] # 添加x轴数据
- for i_num in range(len(iter_np)): # 迭代每一个i(不重复)
- i_data = iter_np[i_num]
- # n_data= i[i == i_data]#取得现在特征数据
- x_axis.append(f'[{i_data}]')
- for num in range(len(class_list)): # 根据类别进行迭代
- now_class = class_list[num] # 取得class_list的布尔数组
- # 切片成和n_data一样的位置一样的形状(now_class就是一个bool矩阵)
- bool_class = now_class[i == i_data]
- # 用len计数 c_list = [[class1的数据],[class2的数据],[]]
- c_list[num][i_num] = (int(np.sum(bool_class).tolist()))
- c = (
- Bar() .add_xaxis(x_axis) .set_global_opts(
- title_opts=opts.TitleOpts(
- title='类型-特征统计柱状图'),
- **global_Set,
- xaxis_opts=opts.AxisOpts(
- type_='category'),
- yaxis_opts=opts.AxisOpts(
- type_='value')))
- y_axis = []
- for i in range(len(c_list)):
- y_axis.append(f'{class_[i]}')
- c.add_yaxis(f'{class_[i]}', c_list[i], **Label_Set)
- desTo_CSV(Dic, f'类型-[{num_i}]特征统计柱状图', c_list, x_axis, y_axis)
- tab.add(c, f'类型-[{num_i}]特征统计柱状图')
- # 未完成
- save = Dic + r'/特征统计.HTML'
- tab.render(save) # 生成HTML
- return save,
- class Numpy_To_HeatMap(To_PyeBase): # Numpy矩阵绘制热力图
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- data = self.x_trainData
- x = [f'横[{i}]' for i in range(len(data))]
- y = [f'纵[{i}]' for i in range(len(data[0]))]
- value = [(f'横[{i}]', f'纵[{j}]', float(data[i][j]))
- for i in range(len(data)) for j in range(len(data[i]))]
- c = (HeatMap()
- .add_xaxis(x)
- .add_yaxis(f'数据', y, value, **Label_Set) # value的第一个数值是x
- .set_global_opts(title_opts=opts.TitleOpts(title='矩阵热力图'), **global_Leg,
- yaxis_opts=opts.AxisOpts(
- is_scale=True, type_='category'), # 'category'
- xaxis_opts=opts.AxisOpts(is_scale=True, type_='category'),
- visualmap_opts=opts.VisualMapOpts(is_show=True, max_=float(data.max()),
- min_=float(data.min()),
- pos_right='3%')) # 显示
- )
- tab.add(c, '矩阵热力图')
- tab.add(make_Tab(x, data.transpose.tolist()), f'矩阵热力图:表格')
- save = Dic + r'/矩阵热力图.HTML'
- tab.render(save) # 生成HTML
- return save,
- class Predictive_HeatMap_Base(To_PyeBase): # 绘制预测型热力图
- def __init__(self, args_use, Learner, *args, **kwargs): # model表示当前选用的模型类型,Alpha针对正则化的参数
- super(
- Predictive_HeatMap_Base,
- self).__init__(
- args_use,
- Learner,
- *
- args,
- **kwargs)
- self.Model = Learner.Model
- self.Select_Model = None
- self.have_Fit = Learner.have_Fit
- self.Model_Name = 'Select_Model'
- self.Learner = Learner
- self.x_trainData = Learner.x_trainData.copy()
- self.y_trainData = Learner.y_trainData.copy()
- self.means = []
- def Fit(self, x_data, *args, **kwargs):
- try:
- self.means = x_data.ravel()
- except BaseException:
- pass
- self.have_Fit = True
- return 'None', 'None'
- def Des(
- self,
- Dic,
- Decision_boundary,
- Prediction_boundary,
- *args,
- **kwargs):
- tab = Tab()
- y = self.y_trainData
- x_data = self.x_trainData
- try: # 如果没有class
- class_ = self.Model.classes_.tolist()
- class_heard = [f'类别[{i}]' for i in range(len(class_))]
- # 获取数据
- get, x_means, x_range, Type = Training_visualization(
- x_data, class_, y)
- # 可使用自带的means,并且nan表示跳过
- for i in range(min([len(x_means), len(self.means)])):
- try:
- g = self.means[i]
- if g == np.nan:
- raise Exception
- x_means[i] = g
- except BaseException:
- pass
- get = Decision_boundary(
- x_range, x_means, self.Learner.predict, class_, Type)
- for i in range(len(get)):
- tab.add(get[i], f'{i}预测热力图')
- heard = class_heard + [f'普适预测第{i}特征' for i in range(len(x_means))]
- data = class_ + [f'{i}' for i in x_means]
- c = Table().add(headers=heard, rows=[data])
- tab.add(c, '数据表')
- except BaseException:
- get, x_means, x_range, Type = regress_visualization(x_data, y)
- get = Prediction_boundary(
- x_range, x_means, self.Learner.predict, Type)
- for i in range(len(get)):
- tab.add(get[i], f'{i}预测热力图')
- heard = [f'普适预测第{i}特征' for i in range(len(x_means))]
- data = [f'{i}' for i in x_means]
- c = Table().add(headers=heard, rows=[data])
- tab.add(c, '数据表')
- save = Dic + r'/预测热力图.HTML'
- tab.render(save) # 生成HTML
- return save,
- class Predictive_HeatMap(Predictive_HeatMap_Base): # 绘制预测型热力图
- def Des(self, Dic, *args, **kwargs):
- return super().Des(Dic, Decision_boundary, Prediction_boundary)
- class Predictive_HeatMap_More(Predictive_HeatMap_Base): # 绘制预测型热力图_More
- def Des(self, Dic, *args, **kwargs):
- return super().Des(Dic, Decision_boundary_More, Prediction_boundary_More)
- class Near_feature_scatter_class_More(To_PyeBase):
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- x_data = self.x_trainData
- y = self.y_trainData
- class_ = np.unique(y).ravel().tolist()
- class_heard = [f'簇[{i}]' for i in range(len(class_))]
- get, x_means, x_range, Type = Training_visualization_More_NoCenter(
- x_data, class_, y)
- for i in range(len(get)):
- tab.add(get[i], f'{i}训练数据散点图')
- heard = class_heard + [f'普适预测第{i}特征' for i in range(len(x_means))]
- data = class_ + [f'{i}' for i in x_means]
- c = Table().add(headers=heard, rows=[data])
- tab.add(c, '数据表')
- save = Dic + r'/数据特征散点图(分类).HTML'
- tab.render(save) # 生成HTML
- return save,
- class Near_feature_scatter_More(To_PyeBase):
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- x_data = self.x_trainData
- x_means = make_Cat(x_data).get()[0]
- get_y = Feature_visualization(x_data, '数据散点图') # 转换
- for i in range(len(get_y)):
- tab.add(get_y[i], f'[{i}]数据x-x散点图')
- heard = [f'普适预测第{i}特征' for i in range(len(x_means))]
- data = [f'{i}' for i in x_means]
- c = Table().add(headers=heard, rows=[data])
- tab.add(c, '数据表')
- save = Dic + r'/数据特征散点图.HTML'
- tab.render(save) # 生成HTML
- return save,
- class Near_feature_scatter_class(To_PyeBase): # 临近特征散点图:分类数据
- def Des(self, Dic, *args, **kwargs):
- # 获取数据
- class_ = np.unique(self.y_trainData).ravel().tolist()
- class_heard = [f'类别[{i}]' for i in range(len(class_))]
- tab = Tab()
- y = self.y_trainData
- x_data = self.x_trainData
- get, x_means, x_range, Type = Training_visualization(x_data, class_, y)
- for i in range(len(get)):
- tab.add(get[i], f'{i}临近特征散点图')
- heard = class_heard + [f'普适预测第{i}特征' for i in range(len(x_means))]
- data = class_ + [f'{i}' for i in x_means]
- c = Table().add(headers=heard, rows=[data])
- tab.add(c, '数据表')
- save = Dic + r'/临近数据特征散点图(分类).HTML'
- tab.render(save) # 生成HTML
- return save,
- class Near_feature_scatter(To_PyeBase): # 临近特征散点图:连续数据
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- x_data = self.x_trainData.transpose
- y = self.y_trainData
- get, x_means, x_range, Type = Training_visualization_NoClass(x_data)
- for i in range(len(get)):
- tab.add(get[i], f'{i}临近特征散点图')
- columns = [f'普适预测第{i}特征' for i in range(len(x_means))]
- data = [f'{i}' for i in x_means]
- tab.add(make_Tab(columns, [data]), '数据表')
- save = Dic + r'/临近数据特征散点图.HTML'
- tab.render(save) # 生成HTML
- return save,
- class Feature_scatter_YX(To_PyeBase): # y-x图
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- x_data = self.x_trainData
- y = self.y_trainData
- get, x_means, x_range, Type = regress_visualization(x_data, y)
- for i in range(len(get)):
- tab.add(get[i], f'{i}特征x-y散点图')
- columns = [f'普适预测第{i}特征' for i in range(len(x_means))]
- data = [f'{i}' for i in x_means]
- tab.add(make_Tab(columns, [data]), '数据表')
- save = Dic + r'/特征y-x图像.HTML'
- tab.render(save) # 生成HTML
- return save,
- class Line_Model(Study_MachineBase):
- def __init__(self, args_use, model, *args, **kwargs): # model表示当前选用的模型类型,Alpha针对正则化的参数
- super(Line_Model, self).__init__(*args, **kwargs)
- Model = {'Line': LinearRegression, 'Ridge': Ridge, 'Lasso': Lasso}[
- model]
- if model == 'Line':
- self.Model = Model()
- self.k = {}
- else:
- self.Model = Model(
- alpha=args_use['alpha'],
- max_iter=args_use['max_iter'])
- self.k = {
- 'alpha': args_use['alpha'],
- 'max_iter': args_use['max_iter']}
- # 记录这两个是为了克隆
- self.Alpha = args_use['alpha']
- self.max_iter = args_use['max_iter']
- self.Model_Name = model
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- x_data = self.x_trainData
- y = self.y_trainData
- w_list = self.Model.coef_.tolist()
- w_heard = [f'系数w[{i}]' for i in range(len(w_list))]
- b = self.Model.intercept_.tolist()
- get, x_means, x_range, Type = regress_visualization(x_data, y)
- get_Line = Regress_W(x_data, y, w_list, b, x_means.copy())
- for i in range(len(get)):
- tab.add(get[i].overlap(get_Line[i]), f'{i}预测类型图')
- get = Prediction_boundary(x_range, x_means, self.Predict, Type)
- for i in range(len(get)):
- tab.add(get[i], f'{i}预测热力图')
- tab.add(scatter(w_heard, w_list), '系数w散点图')
- tab.add(bar(w_heard, self.Model.coef_), '系数柱状图')
- columns = [
- f'普适预测第{i}特征' for i in range(
- len(x_means))] + w_heard + ['截距b']
- data = [f'{i}' for i in x_means] + w_list + [b]
- if self.Model_Name != 'Line':
- columns += ['阿尔法', '最大迭代次数']
- data += [self.Model.alpha, self.Model.max_iter]
- tab.add(make_Tab(columns, [data]), '数据表')
- desTo_CSV(Dic, '系数表', [w_list] +
- [b], [f'系数W[{i}]' for i in range(len(w_list))] +
- ['截距'])
- desTo_CSV(Dic, '预测表', [[f'{i}' for i in x_means]], [
- f'普适预测第{i}特征' for i in range(len(x_means))])
- save = Dic + r'/线性回归模型.HTML'
- tab.render(save) # 生成HTML
- return save,
- class LogisticRegression_Model(Study_MachineBase):
- def __init__(self, args_use, model, *args, **kwargs): # model表示当前选用的模型类型,Alpha针对正则化的参数
- super(LogisticRegression_Model, self).__init__(*args, **kwargs)
- self.Model = LogisticRegression(
- C=args_use['C'], max_iter=args_use['max_iter'])
- # 记录这两个是为了克隆
- self.C = args_use['C']
- self.max_iter = args_use['max_iter']
- self.k = {'C': args_use['C'], 'max_iter': args_use['max_iter']}
- self.Model_Name = model
- def Des(self, Dic='render.html', *args, **kwargs):
- # 获取数据
- w_array = self.Model.coef_
- w_list = w_array.tolist() # 变为表格
- b = self.Model.intercept_
- c = self.Model.C
- max_iter = self.Model.max_iter
- class_ = self.Model.classes_.tolist()
- class_heard = [f'类别[{i}]' for i in range(len(class_))]
- tab = Tab()
- y = self.y_trainData
- x_data = self.x_trainData
- get, x_means, x_range, Type = Training_visualization(x_data, class_, y)
- get_Line = Training_W(x_data, class_, y, w_list, b, x_means.copy())
- for i in range(len(get)):
- tab.add(get[i].overlap(get_Line[i]), f'{i}决策边界散点图')
- for i in range(len(w_list)):
- w = w_list[i]
- w_heard = [f'系数w[{i},{j}]' for j in range(len(w))]
- tab.add(scatter(w_heard, w), f'系数w[{i}]散点图')
- tab.add(bar(w_heard, w_array[i]), f'系数w[{i}]柱状图')
- columns = class_heard + \
- [f'截距{i}' for i in range(len(b))] + ['C', '最大迭代数']
- data = class_ + b.tolist() + [c, max_iter]
- c = Table().add(headers=columns, rows=[data])
- tab.add(c, '数据表')
- c = Table().add(
- headers=[f'系数W[{i}]' for i in range(len(w_list[0]))], rows=w_list)
- tab.add(c, '系数数据表')
- c = Table().add(headers=[f'普适预测第{i}特征' for i in range(
- len(x_means))], rows=[[f'{i}' for i in x_means]])
- tab.add(c, '普适预测数据表')
- desTo_CSV(Dic, '系数表', w_list, [
- f'系数W[{i}]' for i in range(len(w_list[0]))])
- desTo_CSV(Dic, '截距表', [b], [f'截距{i}' for i in range(len(b))])
- desTo_CSV(Dic, '预测表', [[f'{i}' for i in x_means]], [
- f'普适预测第{i}特征' for i in range(len(x_means))])
- save = Dic + r'/逻辑回归.HTML'
- tab.render(save) # 生成HTML
- return save,
- class Categorical_Data: # 数据统计助手
- def __init__(self):
- self.x_means = []
- self.x_range = []
- self.Type = []
- def __call__(self, x1, *args, **kwargs):
- get = self.is_continuous(x1)
- return get
- def is_continuous(self, x1: np.array):
- try:
- x1_con = is_continuous(x1)
- if x1_con:
- self.x_means.append(np.mean(x1))
- self.add_Range(x1)
- else:
- raise Exception
- return x1_con
- except BaseException: # 找出出现次数最多的元素
- new = np.unique(x1) # 去除相同的元素
- count_list = []
- for i in new:
- count_list.append(np.sum(x1 == i))
- index = count_list.index(max(count_list)) # 找出最大值的索引
- self.x_means.append(x1[index])
- self.add_Range(x1, False)
- return False
- def add_Range(self, x1: np.array, range_=True):
- try:
- if not range_:
- raise Exception
- min_ = int(x1.min()) - 1
- max_ = int(x1.max()) + 1
- # 不需要复制列表
- self.x_range.append([min_, max_])
- self.Type.append(1)
- except BaseException:
- self.x_range.append(list(set(x1.tolist()))) # 去除多余元素
- self.Type.append(2)
- def get(self):
- return self.x_means, self.x_range, self.Type
- class Knn_Model(Study_MachineBase):
- def __init__(self, args_use, model, *args, **kwargs): # model表示当前选用的模型类型,Alpha针对正则化的参数
- super(Knn_Model, self).__init__(*args, **kwargs)
- Model = {
- 'Knn_class': KNeighborsClassifier,
- 'Knn': KNeighborsRegressor}[model]
- self.Model = Model(
- p=args_use['p'],
- n_neighbors=args_use['n_neighbors'])
- # 记录这两个是为了克隆
- self.n_neighbors = args_use['n_neighbors']
- self.p = args_use['p']
- self.k = {'n_neighbors': args_use['n_neighbors'], 'p': args_use['p']}
- self.Model_Name = model
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- y = self.y_trainData
- x_data = self.x_trainData
- y_test = self.y_testData
- x_test = self.x_testData
- if self.Model_Name == 'Knn_class':
- class_ = self.Model.classes_.tolist()
- class_heard = [f'类别[{i}]' for i in range(len(class_))]
- get, x_means, x_range, Type = Training_visualization(
- x_data, class_, y)
- for i in range(len(get)):
- tab.add(get[i], f'{i}训练数据散点图')
- if y_test is not None:
- get = Training_visualization(x_test, class_, y_test)[0]
- for i in range(len(get)):
- tab.add(get[i], f'{i}测试数据散点图')
- get = Decision_boundary(
- x_range, x_means, self.Predict, class_, Type)
- for i in range(len(get)):
- tab.add(get[i], f'{i}预测热力图')
- heard = class_heard + [f'普适预测第{i}特征' for i in range(len(x_means))]
- data = class_ + [f'{i}' for i in x_means]
- c = Table().add(headers=heard, rows=[data])
- tab.add(c, '数据表')
- else:
- get, x_means, x_range, Type = regress_visualization(x_data, y)
- for i in range(len(get)):
- tab.add(get[i], f'{i}训练数据散点图')
- get = regress_visualization(x_test, y_test)[0]
- for i in range(len(get)):
- tab.add(get[i], f'{i}测试数据类型图')
- get = Prediction_boundary(x_range, x_means, self.Predict, Type)
- for i in range(len(get)):
- tab.add(get[i], f'{i}预测热力图')
- heard = [f'普适预测第{i}特征' for i in range(len(x_means))]
- data = [f'{i}' for i in x_means]
- c = Table().add(headers=heard, rows=[data])
- tab.add(c, '数据表')
- desTo_CSV(Dic, '预测表', [[f'{i}' for i in x_means]], [
- f'普适预测第{i}特征' for i in range(len(x_means))])
- save = Dic + r'/K.HTML'
- tab.render(save) # 生成HTML
- return save,
- class Tree_Model(Study_MachineBase):
- def __init__(self, args_use, model, *args, **kwargs): # model表示当前选用的模型类型,Alpha针对正则化的参数
- super(Tree_Model, self).__init__(*args, **kwargs)
- Model = {
- 'Tree_class': DecisionTreeClassifier,
- 'Tree': DecisionTreeRegressor}[model]
- self.Model = Model(
- criterion=args_use['criterion'],
- splitter=args_use['splitter'],
- max_features=args_use['max_features'],
- max_depth=args_use['max_depth'],
- min_samples_split=args_use['min_samples_split'])
- # 记录这两个是为了克隆
- self.criterion = args_use['criterion']
- self.splitter = args_use['splitter']
- self.max_features = args_use['max_features']
- self.max_depth = args_use['max_depth']
- self.min_samples_split = args_use['min_samples_split']
- self.k = {
- 'criterion': args_use['criterion'],
- 'splitter': args_use['splitter'],
- 'max_features': args_use['max_features'],
- 'max_depth': args_use['max_depth'],
- 'min_samples_split': args_use['min_samples_split']}
- self.Model_Name = model
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- importance = self.Model.feature_importances_.tolist()
- with open(Dic + r"\Tree_Gra.dot", 'w') as f:
- export_graphviz(self.Model, out_file=f)
- make_bar('特征重要性', importance, tab)
- desTo_CSV(
- Dic, '特征重要性', [importance], [
- f'[{i}]特征' for i in range(
- len(importance))])
- tab.add(SeeTree(Dic + r"\Tree_Gra.dot"), '决策树可视化')
- y = self.y_trainData
- x_data = self.x_trainData
- y_test = self.y_testData
- x_test = self.x_testData
- if self.Model_Name == 'Tree_class':
- class_ = self.Model.classes_.tolist()
- class_heard = [f'类别[{i}]' for i in range(len(class_))]
- get, x_means, x_range, Type = Training_visualization(
- x_data, class_, y)
- for i in range(len(get)):
- tab.add(get[i], f'{i}训练数据散点图')
- get = Training_visualization(x_test, class_, y_test)[0]
- for i in range(len(get)):
- tab.add(get[i], f'{i}测试数据散点图')
- get = Decision_boundary(
- x_range, x_means, self.Predict, class_, Type)
- for i in range(len(get)):
- tab.add(get[i], f'{i}预测热力图')
- tab.add(make_Tab(class_heard +
- [f'普适预测第{i}特征' for i in range(len(x_means))] +
- [f'特征{i}重要性' for i in range(len(importance))], [class_ +
- [f'{i}' for i in x_means] +
- importance]), '数据表')
- else:
- get, x_means, x_range, Type = regress_visualization(x_data, y)
- for i in range(len(get)):
- tab.add(get[i], f'{i}训练数据散点图')
- get = regress_visualization(x_test, y_test)[0]
- for i in range(len(get)):
- tab.add(get[i], f'{i}测试数据类型图')
- get = Prediction_boundary(x_range, x_means, self.Predict, Type)
- for i in range(len(get)):
- tab.add(get[i], f'{i}预测热力图')
- tab.add(make_Tab([f'普适预测第{i}特征' for i in range(len(x_means))] + [f'特征{i}重要性' for i in range(
- len(importance))], [[f'{i}' for i in x_means] + importance]), '数据表')
- desTo_CSV(Dic, '预测表', [[f'{i}' for i in x_means]], [
- f'普适预测第{i}特征' for i in range(len(x_means))])
- save = Dic + r'/决策树.HTML'
- tab.render(save) # 生成HTML
- return save,
- class Forest_Model(Study_MachineBase):
- def __init__(self, args_use, model, *args, **kwargs): # model表示当前选用的模型类型,Alpha针对正则化的参数
- super(Forest_Model, self).__init__(*args, **kwargs)
- Model = {'Forest_class': RandomForestClassifier,
- 'Forest': RandomForestRegressor}[model]
- self.Model = Model(
- n_estimators=args_use['n_Tree'],
- criterion=args_use['criterion'],
- max_features=args_use['max_features'],
- max_depth=args_use['max_depth'],
- min_samples_split=args_use['min_samples_split'])
- # 记录这两个是为了克隆
- self.n_estimators = args_use['n_Tree']
- self.criterion = args_use['criterion']
- self.max_features = args_use['max_features']
- self.max_depth = args_use['max_depth']
- self.min_samples_split = args_use['min_samples_split']
- self.k = {
- 'n_estimators': args_use['n_Tree'],
- 'criterion': args_use['criterion'],
- 'max_features': args_use['max_features'],
- 'max_depth': args_use['max_depth'],
- 'min_samples_split': args_use['min_samples_split']}
- self.Model_Name = model
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- # 多个决策树可视化
- for i in range(len(self.Model.estimators_)):
- with open(Dic + rf"\Tree_Gra[{i}].dot", 'w') as f:
- export_graphviz(self.Model.estimators_[i], out_file=f)
- tab.add(SeeTree(Dic + rf"\Tree_Gra[{i}].dot"), f'[{i}]决策树可视化')
- y = self.y_trainData
- x_data = self.x_trainData
- if self.Model_Name == 'Forest_class':
- class_ = self.Model.classes_.tolist()
- class_heard = [f'类别[{i}]' for i in range(len(class_))]
- get, x_means, x_range, Type = Training_visualization(
- x_data, class_, y)
- for i in range(len(get)):
- tab.add(get[i], f'{i}训练数据散点图')
- get = Decision_boundary(
- x_range, x_means, self.Predict, class_, Type)
- for i in range(len(get)):
- tab.add(get[i], f'{i}预测热力图')
- tab.add(make_Tab(class_heard +
- [f'普适预测第{i}特征' for i in range(len(x_means))], [class_ +
- [f'{i}' for i in x_means]]), '数据表')
- else:
- get, x_means, x_range, Type = regress_visualization(x_data, y)
- for i in range(len(get)):
- tab.add(get[i], f'{i}预测类型图')
- get = Prediction_boundary(x_range, x_means, self.Predict, Type)
- for i in range(len(get)):
- tab.add(get[i], f'{i}预测热力图')
- tab.add(make_Tab([f'普适预测第{i}特征' for i in range(len(x_means))], [
- [f'{i}' for i in x_means]]), '数据表')
- desTo_CSV(Dic, '预测表', [[f'{i}' for i in x_means]], [
- f'普适预测第{i}特征' for i in range(len(x_means))])
- save = Dic + r'/随机森林.HTML'
- tab.render(save) # 生成HTML
- return save,
- class GradientTree_Model(Study_MachineBase): # 继承Tree_Model主要是继承Des
- def __init__(self, args_use, model, *args, **kwargs): # model表示当前选用的模型类型,Alpha针对正则化的参数
- super(GradientTree_Model, self).__init__(
- *args, **kwargs) # 不需要执行Tree_Model的初始化
- Model = {'GradientTree_class': GradientBoostingClassifier,
- 'GradientTree': GradientBoostingRegressor}[model]
- self.Model = Model(
- n_estimators=args_use['n_Tree'],
- max_features=args_use['max_features'],
- max_depth=args_use['max_depth'],
- min_samples_split=args_use['min_samples_split'])
- # 记录这两个是为了克隆
- self.criterion = args_use['criterion']
- self.splitter = args_use['splitter']
- self.max_features = args_use['max_features']
- self.max_depth = args_use['max_depth']
- self.min_samples_split = args_use['min_samples_split']
- self.k = {
- 'criterion': args_use['criterion'],
- 'splitter': args_use['splitter'],
- 'max_features': args_use['max_features'],
- 'max_depth': args_use['max_depth'],
- 'min_samples_split': args_use['min_samples_split']}
- self.Model_Name = model
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- # 多个决策树可视化
- for a in range(len(self.Model.estimators_)):
- for i in range(len(self.Model.estimators_[a])):
- with open(Dic + rf"\Tree_Gra[{a},{i}].dot", 'w') as f:
- export_graphviz(self.Model.estimators_[a][i], out_file=f)
- tab.add(
- SeeTree(
- Dic +
- rf"\Tree_Gra[{a},{i}].dot"),
- f'[{a},{i}]决策树可视化')
- y = self.y_trainData
- x_data = self.x_trainData
- if self.Model_Name == 'Tree_class':
- class_ = self.Model.classes_.tolist()
- class_heard = [f'类别[{i}]' for i in range(len(class_))]
- get, x_means, x_range, Type = Training_visualization(
- x_data, class_, y)
- for i in range(len(get)):
- tab.add(get[i], f'{i}训练数据散点图')
- get = Decision_boundary(
- x_range, x_means, self.Predict, class_, Type)
- for i in range(len(get)):
- tab.add(get[i], f'{i}预测热力图')
- tab.add(make_Tab(class_heard +
- [f'普适预测第{i}特征' for i in range(len(x_means))], [class_ +
- [f'{i}' for i in x_means]]), '数据表')
- else:
- get, x_means, x_range, Type = regress_visualization(x_data, y)
- for i in range(len(get)):
- tab.add(get[i], f'{i}预测类型图')
- get = Prediction_boundary(x_range, x_means, self.Predict, Type)
- for i in range(len(get)):
- tab.add(get[i], f'{i}预测热力图')
- tab.add(make_Tab([f'普适预测第{i}特征' for i in range(len(x_means))], [
- [f'{i}' for i in x_means]]), '数据表')
- desTo_CSV(Dic, '预测表', [[f'{i}' for i in x_means]], [
- f'普适预测第{i}特征' for i in range(len(x_means))])
- save = Dic + r'/梯度提升回归树.HTML'
- tab.render(save) # 生成HTML
- return save,
- class SVC_Model(Study_MachineBase):
- def __init__(self, args_use, model, *args, **kwargs): # model表示当前选用的模型类型,Alpha针对正则化的参数
- super(SVC_Model, self).__init__(*args, **kwargs)
- self.Model = SVC(
- C=args_use['C'],
- gamma=args_use['gamma'],
- kernel=args_use['kernel'])
- # 记录这两个是为了克隆
- self.C = args_use['C']
- self.gamma = args_use['gamma']
- self.kernel = args_use['kernel']
- self.k = {
- 'C': args_use['C'],
- 'gamma': args_use['gamma'],
- 'kernel': args_use['kernel']}
- self.Model_Name = model
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- try:
- w_list = self.Model.coef_.tolist() # 未必有这个属性
- b = self.Model.intercept_.tolist()
- U = True
- except BaseException:
- U = False
- class_ = self.Model.classes_.tolist()
- class_heard = [f'类别[{i}]' for i in range(len(class_))]
- y = self.y_trainData
- x_data = self.x_trainData
- get, x_means, x_range, Type = Training_visualization(x_data, class_, y)
- if U:
- get_Line = Training_W(x_data, class_, y, w_list, b, x_means.copy())
- for i in range(len(get)):
- if U:
- tab.add(get[i].overlap(get_Line[i]), f'{i}决策边界散点图')
- else:
- tab.add(get[i], f'{i}决策边界散点图')
- get = Decision_boundary(x_range, x_means, self.Predict, class_, Type)
- for i in range(len(get)):
- tab.add(get[i], f'{i}预测热力图')
- dic = {2: '离散', 1: '连续'}
- tab.add(make_Tab(class_heard +
- [f'普适预测第{i}特征:{dic[Type[i]]}' for i in range(len(x_means))], [class_ +
- [f'{i}' for i in x_means]]), '数据表')
- if U:
- desTo_CSV(Dic, '系数表', w_list, [
- f'系数W[{i}]' for i in range(len(w_list[0]))])
- if U:
- desTo_CSV(Dic, '截距表', [b], [f'截距{i}' for i in range(len(b))])
- desTo_CSV(Dic, '预测表', [[f'{i}' for i in x_means]], [
- f'普适预测第{i}特征' for i in range(len(x_means))])
- save = Dic + r'/支持向量机分类.HTML'
- tab.render(save) # 生成HTML
- return save,
- class SVR_Model(Study_MachineBase):
- def __init__(self, args_use, model, *args, **kwargs): # model表示当前选用的模型类型,Alpha针对正则化的参数
- super(SVR_Model, self).__init__(*args, **kwargs)
- self.Model = SVR(
- C=args_use['C'],
- gamma=args_use['gamma'],
- kernel=args_use['kernel'])
- # 记录这两个是为了克隆
- self.C = args_use['C']
- self.gamma = args_use['gamma']
- self.kernel = args_use['kernel']
- self.k = {
- 'C': args_use['C'],
- 'gamma': args_use['gamma'],
- 'kernel': args_use['kernel']}
- self.Model_Name = model
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- x_data = self.x_trainData
- y = self.y_trainData
- try:
- w_list = self.Model.coef_.tolist() # 未必有这个属性
- b = self.Model.intercept_.tolist()
- U = True
- except BaseException:
- U = False
- get, x_means, x_range, Type = regress_visualization(x_data, y)
- if U:
- get_Line = Regress_W(x_data, y, w_list, b, x_means.copy())
- for i in range(len(get)):
- if U:
- tab.add(get[i].overlap(get_Line[i]), f'{i}预测类型图')
- else:
- tab.add(get[i], f'{i}预测类型图')
- get = Prediction_boundary(x_range, x_means, self.Predict, Type)
- for i in range(len(get)):
- tab.add(get[i], f'{i}预测热力图')
- if U:
- desTo_CSV(Dic, '系数表', w_list, [
- f'系数W[{i}]' for i in range(len(w_list[0]))])
- if U:
- desTo_CSV(Dic, '截距表', [b], [f'截距{i}' for i in range(len(b))])
- desTo_CSV(Dic, '预测表', [[f'{i}' for i in x_means]], [
- f'普适预测第{i}特征' for i in range(len(x_means))])
- tab.add(make_Tab([f'普适预测第{i}特征' for i in range(len(x_means))], [
- [f'{i}' for i in x_means]]), '数据表')
- save = Dic + r'/支持向量机回归.HTML'
- tab.render(save) # 生成HTML
- return save,
- class Variance_Model(Unsupervised): # 无监督
- def __init__(self, args_use, model, *args, **kwargs): # model表示当前选用的模型类型,Alpha针对正则化的参数
- super(Variance_Model, self).__init__(*args, **kwargs)
- self.Model = VarianceThreshold(
- threshold=(args_use['P'] * (1 - args_use['P'])))
- # 记录这两个是为了克隆
- self.threshold = args_use['P']
- self.k = {'threshold': args_use['P']}
- self.Model_Name = model
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- var = self.Model.variances_ # 标准差
- y_data = self.y_testData
- if isinstance(y_data, np.ndarray):
- get = Feature_visualization(self.y_testData)
- for i in range(len(get)):
- tab.add(get[i], f'[{i}]数据x-x散点图')
- c = (
- Bar()
- .add_xaxis([f'[{i}]特征' for i in range(len(var))])
- .add_yaxis('标准差', var.tolist(), **Label_Set)
- .set_global_opts(title_opts=opts.TitleOpts(title='系数w柱状图'), **global_Set)
- )
- tab.add(c, '数据标准差')
- save = Dic + r'/方差特征选择.HTML'
- tab.render(save) # 生成HTML
- return save,
- class SelectKBest_Model(prep_Base): # 有监督
- def __init__(self, args_use, model, *args, **kwargs):
- super(SelectKBest_Model, self).__init__(*args, **kwargs)
- self.Model = SelectKBest(
- k=args_use['k'],
- score_func=args_use['score_func'])
- # 记录这两个是为了克隆
- self.k_ = args_use['k']
- self.score_func = args_use['score_func']
- self.k = {'k': args_use['k'], 'score_func': args_use['score_func']}
- self.Model_Name = model
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- score = self.Model.scores_.tolist()
- support = self.Model.get_support()
- y_data = self.y_trainData
- x_data = self.x_trainData
- if isinstance(x_data, np.ndarray):
- get = Feature_visualization(x_data)
- for i in range(len(get)):
- tab.add(get[i], f'[{i}]训练数据x-x散点图')
- if isinstance(y_data, np.ndarray):
- get = Feature_visualization(y_data)
- for i in range(len(get)):
- tab.add(get[i], f'[{i}]保留训练数据x-x散点图')
- y_data = self.y_testData
- x_data = self.x_testData
- if isinstance(x_data, np.ndarray):
- get = Feature_visualization(x_data)
- for i in range(len(get)):
- tab.add(get[i], f'[{i}]数据x-x散点图')
- if isinstance(y_data, np.ndarray):
- get = Feature_visualization(y_data)
- for i in range(len(get)):
- tab.add(get[i], f'[{i}]保留数据x-x散点图')
- Choose = []
- UnChoose = []
- for i in range(len(score)):
- if support[i]:
- Choose.append(score[i])
- UnChoose.append(0) # 占位
- else:
- UnChoose.append(score[i])
- Choose.append(0)
- c = (
- Bar()
- .add_xaxis([f'[{i}]特征' for i in range(len(score))])
- .add_yaxis('选中特征', Choose, **Label_Set)
- .add_yaxis('抛弃特征', UnChoose, **Label_Set)
- .set_global_opts(title_opts=opts.TitleOpts(title='系数w柱状图'), **global_Set)
- )
- tab.add(c, '单变量重要程度')
- save = Dic + r'/单一变量特征选择.HTML'
- tab.render(save) # 生成HTML
- return save,
- class SelectFrom_Model(prep_Base): # 有监督
- def __init__(self, args_use, Learner, *args, **kwargs): # model表示当前选用的模型类型,Alpha针对正则化的参数
- super(SelectFrom_Model, self).__init__(*args, **kwargs)
- self.Model = Learner.Model
- self.Select_Model = SelectFromModel(
- estimator=Learner.Model,
- max_features=args_use['k'],
- prefit=Learner.have_Fit)
- self.max_features = args_use['k']
- self.estimator = Learner.Model
- self.k = {
- 'max_features': args_use['k'],
- 'estimator': Learner.Model,
- 'have_Fit': Learner.have_Fit}
- self.have_Fit = Learner.have_Fit
- self.Model_Name = 'SelectFrom_Model'
- self.Learner = Learner
- def Fit(self, x_data, y_data, split=0.3, *args, **kwargs):
- y_data = y_data.ravel()
- if not self.have_Fit: # 不允许第二次训练
- self.Select_Model.fit(x_data, y_data)
- self.have_Fit = True
- return 'None', 'None'
- def Predict(self, x_data, *args, **kwargs):
- try:
- self.x_testData = x_data.copy()
- x_Predict = self.Select_Model.transform(x_data)
- self.y_testData = x_Predict.copy()
- self.have_Predict = True
- return x_Predict, '模型特征工程'
- except BaseException:
- self.have_Predict = True
- return np.array([]), '无结果工程'
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- support = self.Select_Model.get_support()
- y_data = self.y_testData
- x_data = self.x_testData
- if isinstance(x_data, np.ndarray):
- get = Feature_visualization(x_data)
- for i in range(len(get)):
- tab.add(get[i], f'[{i}]数据x-x散点图')
- if isinstance(y_data, np.ndarray):
- get = Feature_visualization(y_data)
- for i in range(len(get)):
- tab.add(get[i], f'[{i}]保留数据x-x散点图')
- def make_Bar(score):
- Choose = []
- UnChoose = []
- for i in range(len(score)):
- if support[i]:
- Choose.append(abs(score[i]))
- UnChoose.append(0) # 占位
- else:
- UnChoose.append(abs(score[i]))
- Choose.append(0)
- c = (
- Bar()
- .add_xaxis([f'[{i}]特征' for i in range(len(score))])
- .add_yaxis('选中特征', Choose, **Label_Set)
- .add_yaxis('抛弃特征', UnChoose, **Label_Set)
- .set_global_opts(title_opts=opts.TitleOpts(title='系数w柱状图'), **global_Set)
- )
- tab.add(c, '单变量重要程度')
- try:
- make_Bar(self.Model.coef_)
- except BaseException:
- try:
- make_Bar(self.Model.feature_importances_)
- except BaseException:
- pass
- save = Dic + r'/模型特征选择.HTML'
- tab.render(save) # 生成HTML
- return save,
- class Standardization_Model(Unsupervised): # z-score标准化 无监督
- def __init__(self, args_use, model, *args, **kwargs):
- super(Standardization_Model, self).__init__(*args, **kwargs)
- self.Model = StandardScaler()
- self.k = {}
- self.Model_Name = 'StandardScaler'
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- y_data = self.y_testData
- x_data = self.x_testData
- var = self.Model.var_.tolist()
- means = self.Model.mean_.tolist()
- scale = self.Model.scale_.tolist()
- Conversion_control(y_data, x_data, tab)
- make_bar('标准差', var, tab)
- make_bar('方差', means, tab)
- make_bar('Scale', scale, tab)
- save = Dic + r'/z-score标准化.HTML'
- tab.render(save) # 生成HTML
- return save,
- class MinMaxScaler_Model(Unsupervised): # 离差标准化
- def __init__(self, args_use, model, *args, **kwargs):
- super(MinMaxScaler_Model, self).__init__(*args, **kwargs)
- self.Model = MinMaxScaler(feature_range=args_use['feature_range'])
- self.k = {}
- self.Model_Name = 'MinMaxScaler'
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- y_data = self.y_testData
- x_data = self.x_testData
- scale = self.Model.scale_.tolist()
- max_ = self.Model.data_max_.tolist()
- min_ = self.Model.data_min_.tolist()
- Conversion_control(y_data, x_data, tab)
- make_bar('Scale', scale, tab)
- tab.add(
- make_Tab(
- heard=[
- f'[{i}]特征最大值' for i in range(
- len(max_))] + [
- f'[{i}]特征最小值' for i in range(
- len(min_))], row=[
- max_ + min_]), '数据表格')
- save = Dic + r'/离差标准化.HTML'
- tab.render(save) # 生成HTML
- return save,
- class LogScaler_Model(prep_Base): # 对数标准化
- def __init__(self, args_use, model, *args, **kwargs):
- super(LogScaler_Model, self).__init__(*args, **kwargs)
- self.Model = None
- self.k = {}
- self.Model_Name = 'LogScaler'
- def Fit(self, x_data, *args, **kwargs):
- if not self.have_Predict: # 不允许第二次训练
- self.max_logx = np.log(x_data.max())
- self.have_Fit = True
- return 'None', 'None'
- def Predict(self, x_data, *args, **kwargs):
- try:
- max_logx = self.max_logx
- except BaseException:
- self.have_Fit = False
- self.Fit(x_data)
- max_logx = self.max_logx
- self.x_testData = x_data.copy()
- x_Predict = (np.log(x_data) / max_logx)
- self.y_testData = x_Predict.copy()
- self.have_Predict = True
- return x_Predict, '对数变换'
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- y_data = self.y_testData
- x_data = self.x_testData
- Conversion_control(y_data, x_data, tab)
- tab.add(make_Tab(heard=['最大对数值(自然对数)'],
- row=[[str(self.max_logx)]]), '数据表格')
- save = Dic + r'/对数标准化.HTML'
- tab.render(save) # 生成HTML
- return save,
- class atanScaler_Model(prep_Base): # atan标准化
- def __init__(self, args_use, model, *args, **kwargs):
- super(atanScaler_Model, self).__init__(*args, **kwargs)
- self.Model = None
- self.k = {}
- self.Model_Name = 'atanScaler'
- def Fit(self, x_data, *args, **kwargs):
- self.have_Fit = True
- return 'None', 'None'
- def Predict(self, x_data, *args, **kwargs):
- self.x_testData = x_data.copy()
- x_Predict = (np.arctan(x_data) * (2 / np.pi))
- self.y_testData = x_Predict.copy()
- self.have_Predict = True
- return x_Predict, 'atan变换'
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- y_data = self.y_testData
- x_data = self.x_testData
- Conversion_control(y_data, x_data, tab)
- save = Dic + r'/反正切函数标准化.HTML'
- tab.render(save) # 生成HTML
- return save,
- class decimalScaler_Model(prep_Base): # 小数定标准化
- def __init__(self, args_use, model, *args, **kwargs):
- super(decimalScaler_Model, self).__init__(*args, **kwargs)
- self.Model = None
- self.k = {}
- self.Model_Name = 'Decimal_normalization'
- def Fit(self, x_data, *args, **kwargs):
- if not self.have_Predict: # 不允许第二次训练
- self.j = max([judging_Digits(x_data.max()),
- judging_Digits(x_data.min())])
- self.have_Fit = True
- return 'None', 'None'
- def Predict(self, x_data, *args, **kwargs):
- self.x_testData = x_data.copy()
- try:
- j = self.j
- except BaseException:
- self.have_Fit = False
- self.Fit(x_data)
- j = self.j
- x_Predict = (x_data / (10**j))
- self.y_testData = x_Predict.copy()
- self.have_Predict = True
- return x_Predict, '小数定标标准化'
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- y_data = self.y_testData
- x_data = self.x_testData
- j = self.j
- Conversion_control(y_data, x_data, tab)
- tab.add(make_Tab(heard=['小数位数:j'], row=[[j]]), '数据表格')
- save = Dic + r'/小数定标标准化.HTML'
- tab.render(save) # 生成HTML
- return save,
- class Mapzoom_Model(prep_Base): # 映射标准化
- def __init__(self, args_use, model, *args, **kwargs):
- super(Mapzoom_Model, self).__init__(*args, **kwargs)
- self.Model = None
- self.feature_range = args_use['feature_range']
- self.k = {}
- self.Model_Name = 'Decimal_normalization'
- def Fit(self, x_data, *args, **kwargs):
- if not self.have_Predict: # 不允许第二次训练
- self.max = x_data.max()
- self.min = x_data.min()
- self.have_Fit = True
- return 'None', 'None'
- def Predict(self, x_data, *args, **kwargs):
- self.x_testData = x_data.copy()
- try:
- max = self.max
- min = self.min
- except BaseException:
- self.have_Fit = False
- self.Fit(x_data)
- max = self.max
- min = self.min
- x_Predict = (
- x_data * (self.feature_range[1] - self.feature_range[0])) / (max - min)
- self.y_testData = x_Predict.copy()
- self.have_Predict = True
- return x_Predict, '映射标准化'
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- y_data = self.y_testData
- x_data = self.x_testData
- max = self.max
- min = self.min
- Conversion_control(y_data, x_data, tab)
- tab.add(make_Tab(heard=['最大值', '最小值'], row=[[max, min]]), '数据表格')
- save = Dic + r'/映射标准化.HTML'
- tab.render(save) # 生成HTML
- return save,
- class sigmodScaler_Model(prep_Base): # sigmod变换
- def __init__(self, args_use, model, *args, **kwargs):
- super(sigmodScaler_Model, self).__init__(*args, **kwargs)
- self.Model = None
- self.k = {}
- self.Model_Name = 'sigmodScaler_Model'
- def Fit(self, x_data, *args, **kwargs):
- self.have_Fit = True
- return 'None', 'None'
- def Predict(self, x_data: np.array, *args, **kwargs):
- self.x_testData = x_data.copy()
- x_Predict = (1 / (1 + np.exp(-x_data)))
- self.y_testData = x_Predict.copy()
- self.have_Predict = True
- return x_Predict, 'Sigmod变换'
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- y_data = self.y_testData
- x_data = self.x_testData
- Conversion_control(y_data, x_data, tab)
- save = Dic + r'/Sigmoid变换.HTML'
- tab.render(save) # 生成HTML
- return save,
- class Fuzzy_quantization_Model(prep_Base): # 模糊量化标准化
- def __init__(self, args_use, model, *args, **kwargs):
- super(Fuzzy_quantization_Model, self).__init__(*args, **kwargs)
- self.Model = None
- self.feature_range = args_use['feature_range']
- self.k = {}
- self.Model_Name = 'Fuzzy_quantization'
- def Fit(self, x_data, *args, **kwargs):
- if not self.have_Predict: # 不允许第二次训练
- self.max = x_data.max()
- self.min = x_data.min()
- self.have_Fit = True
- return 'None', 'None'
- def Predict(self, x_data, *args, **kwargs):
- self.x_testData = x_data.copy()
- try:
- max = self.max
- min = self.min
- except BaseException:
- self.have_Fit = False
- self.Fit(x_data)
- max = self.max
- min = self.min
- x_Predict = 1 / 2 + (1 / 2) * np.sin(np.pi / \
- (max - min) * (x_data - (max - min) / 2))
- self.y_testData = x_Predict.copy()
- self.have_Predict = True
- return x_Predict, '模糊量化标准化'
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- y_data = self.y_trainData
- x_data = self.x_trainData
- max = self.max
- min = self.min
- Conversion_control(y_data, x_data, tab)
- tab.add(make_Tab(heard=['最大值', '最小值'], row=[[max, min]]), '数据表格')
- save = Dic + r'/模糊量化标准化.HTML'
- tab.render(save) # 生成HTML
- return save,
- class Regularization_Model(Unsupervised): # 正则化
- def __init__(self, args_use, model, *args, **kwargs):
- super(Regularization_Model, self).__init__(*args, **kwargs)
- self.Model = Normalizer(norm=args_use['norm'])
- self.k = {'norm': args_use['norm']}
- self.Model_Name = 'Regularization'
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- y_data = self.y_testData.copy()
- x_data = self.x_testData.copy()
- Conversion_control(y_data, x_data, tab)
- save = Dic + r'/正则化.HTML'
- tab.render(save) # 生成HTML
- return save,
- # 离散数据
- class Binarizer_Model(Unsupervised): # 二值化
- def __init__(self, args_use, model, *args, **kwargs):
- super(Binarizer_Model, self).__init__(*args, **kwargs)
- self.Model = Binarizer(threshold=args_use['threshold'])
- self.k = {}
- self.Model_Name = 'Binarizer'
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- y_data = self.y_testData
- x_data = self.x_testData
- get_y = Discrete_Feature_visualization(y_data, '转换数据') # 转换
- for i in range(len(get_y)):
- tab.add(get_y[i], f'[{i}]数据x-x离散散点图')
- heard = [f'特征:{i}' for i in range(len(x_data[0]))]
- tab.add(make_Tab(heard, x_data.tolist()), f'原数据')
- tab.add(make_Tab(heard, y_data.tolist()), f'编码数据')
- tab.add(
- make_Tab(
- heard, np.dstack(
- (x_data, y_data)).tolist()), f'合成[原数据,编码]数据')
- save = Dic + r'/二值离散化.HTML'
- tab.render(save) # 生成HTML
- return save,
- class Discretization_Model(prep_Base): # n值离散
- def __init__(self, args_use, model, *args, **kwargs):
- super(Discretization_Model, self).__init__(*args, **kwargs)
- self.Model = None
- range_ = args_use['split_range']
- if range_ == []:
- raise Exception
- elif len(range_) == 1:
- range_.append(range_[0])
- self.range = range_
- self.k = {}
- self.Model_Name = 'Discretization'
- def Fit(self, *args, **kwargs):
- # t值在模型创建时已经保存
- self.have_Fit = True
- return 'None', 'None'
- def Predict(self, x_data, *args, **kwargs):
- self.x_testData = x_data.copy()
- x_Predict = x_data.copy() # 复制
- range_ = self.range
- bool_list = []
- max_ = len(range_) - 1
- o_t = None
- for i in range(len(range_)):
- try:
- t = float(range_[i])
- except BaseException:
- continue
- if o_t is None: # 第一个参数
- bool_list.append(x_Predict <= t)
- else:
- bool_list.append((o_t <= x_Predict) == (x_Predict < t))
- if i == max_:
- bool_list.append(t <= x_Predict)
- o_t = t
- for i in range(len(bool_list)):
- x_Predict[bool_list[i]] = i
- self.y_testData = x_Predict.copy()
- self.have_Predict = True
- return x_Predict, f'{len(bool_list)}值离散化'
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- y_data = self.y_testData
- x_data = self.x_testData
- get_y = Discrete_Feature_visualization(y_data, '转换数据') # 转换
- for i in range(len(get_y)):
- tab.add(get_y[i], f'[{i}]数据x-x离散散点图')
- heard = [f'特征:{i}' for i in range(len(x_data[0]))]
- tab.add(make_Tab(heard, x_data.tolist()), f'原数据')
- tab.add(make_Tab(heard, y_data.tolist()), f'编码数据')
- tab.add(
- make_Tab(
- heard, np.dstack(
- (x_data, y_data)).tolist()), f'合成[原数据,编码]数据')
- save = Dic + r'/多值离散化.HTML'
- tab.render(save) # 生成HTML
- return save,
- class Label_Model(prep_Base): # 数字编码
- def __init__(self, args_use, model, *args, **kwargs):
- super(Label_Model, self).__init__(*args, **kwargs)
- self.Model = []
- self.k = {}
- self.Model_Name = 'LabelEncoder'
- def Fit(self, x_data, *args, **kwargs):
- if not self.have_Predict: # 不允许第二次训练
- self.Model = []
- if x_data.ndim == 1:
- x_data = np.array([x_data])
- for i in range(x_data.shape[1]):
- self.Model.append(LabelEncoder().fit(
- np.ravel(x_data[:, i]))) # 训练机器(每个特征一个学习器)
- self.have_Fit = True
- return 'None', 'None'
- def Predict(self, x_data, *args, **kwargs):
- self.x_testData = x_data.copy()
- x_Predict = x_data.copy()
- if x_data.ndim == 1:
- x_data = np.array([x_data])
- for i in range(x_data.shape[1]):
- x_Predict[:, i] = self.Model[i].transform(x_data[:, i])
- self.y_testData = x_Predict.copy()
- self.have_Predict = True
- return x_Predict, '数字编码'
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- x_data = self.x_testData
- y_data = self.y_testData
- get_y = Discrete_Feature_visualization(y_data, '转换数据') # 转换
- for i in range(len(get_y)):
- tab.add(get_y[i], f'[{i}]数据x-x离散散点图')
- heard = [f'特征:{i}' for i in range(len(x_data[0]))]
- tab.add(make_Tab(heard, x_data.tolist()), f'原数据')
- tab.add(make_Tab(heard, y_data.tolist()), f'编码数据')
- tab.add(
- make_Tab(
- heard, np.dstack(
- (x_data, y_data)).tolist()), f'合成[原数据,编码]数据')
- save = Dic + r'/数字编码.HTML'
- tab.render(save) # 生成HTML
- return save,
- class OneHotEncoder_Model(prep_Base): # 独热编码
- def __init__(self, args_use, model, *args, **kwargs):
- super(OneHotEncoder_Model, self).__init__(*args, **kwargs)
- self.Model = []
- self.ndim_up = args_use['ndim_up']
- self.k = {}
- self.Model_Name = 'OneHotEncoder'
- self.OneHot_Data = None # 三维独热编码
- def Fit(self, x_data, *args, **kwargs):
- if not self.have_Predict: # 不允许第二次训练
- if x_data.ndim == 1:
- x_data = [x_data]
- for i in range(x_data.shape[1]):
- data = np.expand_dims(x_data[:, i], axis=1) # 独热编码需要升维
- self.Model.append(OneHotEncoder().fit(data)) # 训练机器
- self.have_Fit = True
- return 'None', 'None'
- def Predict(self, x_data, *args, **kwargs):
- self.x_testData = x_data.copy()
- x_new = []
- for i in range(x_data.shape[1]):
- data = np.expand_dims(x_data[:, i], axis=1) # 独热编码需要升维
- oneHot = self.Model[i].transform(data).toarray().tolist()
- x_new.append(oneHot) # 添加到列表中
- # 新列表的行数据是原data列数据的独热码(只需要ndim=2,暂时没想到numpy的做法)
- x_new = np.array(x_new)
- x_Predict = []
- for i in range(x_new.shape[1]):
- x_Predict.append(x_new[:, i])
- x_Predict = np.array(x_Predict) # 转换回array
- self.OneHot_Data = x_Predict.copy() # 保存未降维数据
- if not self.ndim_up: # 压缩操作
- new_xPredict = []
- for i in x_Predict:
- new_list = []
- list_ = i.tolist()
- for a in list_:
- new_list += a
- new = np.array(new_list)
- new_xPredict.append(new)
- self.y_testData = np.array(new_xPredict)
- return self.y_testData.copy(), '独热编码'
- self.y_testData = self.OneHot_Data
- self.have_Predict = True
- return x_Predict, '独热编码'
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- y_data = self.y_testData
- x_data = self.x_testData
- oh_data = self.OneHot_Data
- if not self.ndim_up:
- get_y = Discrete_Feature_visualization(y_data, '转换数据') # 转换
- for i in range(len(get_y)):
- tab.add(get_y[i], f'[{i}]数据x-x离散散点图')
- heard = [f'特征:{i}' for i in range(len(x_data[0]))]
- tab.add(make_Tab(heard, x_data.tolist()), f'原数据')
- tab.add(make_Tab(heard, oh_data.tolist()), f'编码数据')
- tab.add(
- make_Tab(
- heard, np.dstack(
- (oh_data, x_data)).tolist()), f'合成[原数据,编码]数据')
- tab.add(make_Tab([f'编码:{i}' for i in range(
- len(y_data[0]))], y_data.tolist()), f'数据')
- save = Dic + r'/独热编码.HTML'
- tab.render(save) # 生成HTML
- return save,
- class Missed_Model(Unsupervised): # 缺失数据补充
- def __init__(self, args_use, model, *args, **kwargs):
- super(Missed_Model, self).__init__(*args, **kwargs)
- self.Model = SimpleImputer(
- missing_values=args_use['miss_value'],
- strategy=args_use['fill_method'],
- fill_value=args_use['fill_value'])
- self.k = {}
- self.Model_Name = 'Missed'
- def Predict(self, x_data, *args, **kwargs):
- self.x_testData = x_data.copy()
- x_Predict = self.Model.transform(x_data)
- self.y_testData = x_Predict.copy()
- self.have_Predict = True
- return x_Predict, '填充缺失'
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- y_data = self.y_testData
- x_data = self.x_testData
- statistics = self.Model.statistics_.tolist()
- Conversion_control(y_data, x_data, tab)
- tab.add(make_Tab([f'特征[{i}]' for i in range(
- len(statistics))], [statistics]), '填充值')
- save = Dic + r'/缺失数据填充.HTML'
- tab.render(save) # 生成HTML
- return save,
- class PCA_Model(Unsupervised):
- def __init__(self, args_use, model, *args, **kwargs):
- super(PCA_Model, self).__init__(*args, **kwargs)
- self.Model = PCA(
- n_components=args_use['n_components'],
- whiten=args_use['white_PCA'])
- self.whiten = args_use['white_PCA']
- self.n_components = args_use['n_components']
- self.k = {
- 'n_components': args_use['n_components'],
- 'whiten': args_use['white_PCA']}
- self.Model_Name = 'PCA'
- def Predict(self, x_data, *args, **kwargs):
- self.x_testData = x_data.copy()
- x_Predict = self.Model.transform(x_data)
- self.y_testData = x_Predict.copy()
- self.have_Predict = True
- return x_Predict, 'PCA'
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- y_data = self.y_testData
- importance = self.Model.components_.tolist()
- var = self.Model.explained_variance_.tolist() # 方量差
- Conversion_Separate_Format(y_data, tab)
- x_data = [f'第{i+1}主成分' for i in range(len(importance))] # 主成分
- y_data = [f'特征[{i}]' for i in range(len(importance[0]))] # 主成分
- value = [(f'第{i+1}主成分', f'特征[{j}]', importance[i][j])
- for i in range(len(importance)) for j in range(len(importance[i]))]
- c = (HeatMap()
- .add_xaxis(x_data)
- .add_yaxis(f'', y_data, value, **Label_Set) # value的第一个数值是x
- .set_global_opts(title_opts=opts.TitleOpts(title='预测热力图'), **global_Leg,
- yaxis_opts=opts.AxisOpts(
- is_scale=True), # 'category'
- xaxis_opts=opts.AxisOpts(is_scale=True),
- visualmap_opts=opts.VisualMapOpts(is_show=True, max_=int(self.Model.components_.max()) + 1,
- min_=int(
- self.Model.components_.min()),
- pos_right='3%')) # 显示
- )
- tab.add(c, '成分热力图')
- c = (
- Bar()
- .add_xaxis([f'第[{i}]主成分' for i in range(len(var))])
- .add_yaxis('方量差', var, **Label_Set)
- .set_global_opts(title_opts=opts.TitleOpts(title='方量差柱状图'), **global_Set)
- )
- desTo_CSV(Dic, '成分重要性', importance, [x_data], [y_data])
- desTo_CSV(Dic, '方量差', [var], [f'第[{i}]主成分' for i in range(len(var))])
- tab.add(c, '方量差柱状图')
- save = Dic + r'/主成分分析.HTML'
- tab.render(save) # 生成HTML
- return save,
- class RPCA_Model(Unsupervised):
- def __init__(self, args_use, model, *args, **kwargs):
- super(RPCA_Model, self).__init__(*args, **kwargs)
- self.Model = IncrementalPCA(
- n_components=args_use['n_components'],
- whiten=args_use['white_PCA'])
- self.n_components = args_use['n_components']
- self.whiten = args_use['white_PCA']
- self.k = {
- 'n_components': args_use['n_components'],
- 'whiten': args_use['white_PCA']}
- self.Model_Name = 'RPCA'
- def Predict(self, x_data, *args, **kwargs):
- self.x_testData = x_data.copy()
- x_Predict = self.Model.transform(x_data)
- self.y_testData = x_Predict.copy()
- self.have_Predict = True
- return x_Predict, 'RPCA'
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- y_data = self.y_trainData
- importance = self.Model.components_.tolist()
- var = self.Model.explained_variance_.tolist() # 方量差
- Conversion_Separate_Format(y_data, tab)
- x_data = [f'第{i + 1}主成分' for i in range(len(importance))] # 主成分
- y_data = [f'特征[{i}]' for i in range(len(importance[0]))] # 主成分
- value = [(f'第{i + 1}主成分', f'特征[{j}]', importance[i][j])
- for i in range(len(importance)) for j in range(len(importance[i]))]
- c = (HeatMap()
- .add_xaxis(x_data)
- .add_yaxis(f'', y_data, value, **Label_Set) # value的第一个数值是x
- .set_global_opts(title_opts=opts.TitleOpts(title='预测热力图'), **global_Leg,
- yaxis_opts=opts.AxisOpts(
- is_scale=True), # 'category'
- xaxis_opts=opts.AxisOpts(is_scale=True),
- visualmap_opts=opts.VisualMapOpts(is_show=True,
- max_=int(
- self.Model.components_.max()) + 1,
- min_=int(
- self.Model.components_.min()),
- pos_right='3%')) # 显示
- )
- tab.add(c, '成分热力图')
- c = (
- Bar()
- .add_xaxis([f'第[{i}]主成分' for i in range(len(var))])
- .add_yaxis('放量差', var, **Label_Set)
- .set_global_opts(title_opts=opts.TitleOpts(title='方量差柱状图'), **global_Set)
- )
- tab.add(c, '方量差柱状图')
- desTo_CSV(Dic, '成分重要性', importance, [x_data], [y_data])
- desTo_CSV(Dic, '方量差', [var], [f'第[{i}]主成分' for i in range(len(var))])
- save = Dic + r'/RPCA(主成分分析).HTML'
- tab.render(save) # 生成HTML
- return save,
- class KPCA_Model(Unsupervised):
- def __init__(self, args_use, model, *args, **kwargs):
- super(KPCA_Model, self).__init__(*args, **kwargs)
- self.Model = KernelPCA(
- n_components=args_use['n_components'],
- kernel=args_use['kernel'])
- self.n_components = args_use['n_components']
- self.kernel = args_use['kernel']
- self.k = {
- 'n_components': args_use['n_components'],
- 'kernel': args_use['kernel']}
- self.Model_Name = 'KPCA'
- def Predict(self, x_data, *args, **kwargs):
- self.x_testData = x_data.copy()
- x_Predict = self.Model.transform(x_data)
- self.y_testData = x_Predict.copy()
- self.have_Predict = True
- return x_Predict, 'KPCA'
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- y_data = self.y_testData
- Conversion_Separate_Format(y_data, tab)
- save = Dic + r'/KPCA(主成分分析).HTML'
- tab.render(save) # 生成HTML
- return save,
- class LDA_Model(prep_Base): # 有监督学习
- def __init__(self, args_use, model, *args, **kwargs):
- super(LDA_Model, self).__init__(*args, **kwargs)
- self.Model = LDA(n_components=args_use['n_components'])
- self.n_components = args_use['n_components']
- self.k = {'n_components': args_use['n_components']}
- self.Model_Name = 'LDA'
- def Predict(self, x_data, *args, **kwargs):
- self.x_testData = x_data.copy()
- x_Predict = self.Model.transform(x_data)
- self.y_testData = x_Predict.copy()
- self.have_Predict = True
- return x_Predict, 'LDA'
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- x_data = self.x_testData
- y_data = self.y_testData
- Conversion_Separate_Format(y_data, tab)
- w_list = self.Model.coef_.tolist() # 变为表格
- b = self.Model.intercept_
- tab = Tab()
- x_means = make_Cat(x_data).get()[0]
- # 回归的y是历史遗留问题 不用分类回归:因为得不到分类数据(predict结果是降维数据不是预测数据)
- get = Regress_W(x_data, None, w_list, b, x_means.copy())
- for i in range(len(get)):
- tab.add(get[i].overlap(get[i]), f'类别:{i}LDA映射曲线')
- save = Dic + r'/render.HTML'
- tab.render(save) # 生成HTML
- return save,
- class NMF_Model(Unsupervised):
- def __init__(self, args_use, model, *args, **kwargs):
- super(NMF_Model, self).__init__(*args, **kwargs)
- self.Model = NMF(n_components=args_use['n_components'])
- self.n_components = args_use['n_components']
- self.k = {'n_components': args_use['n_components']}
- self.Model_Name = 'NFM'
- self.h_testData = None
- # x_trainData保存的是W,h_trainData和y_trainData是后来数据
- def Predict(self, x_data, x_name='', Add_Func=None, *args, **kwargs):
- self.x_testData = x_data.copy()
- x_Predict = self.Model.transform(x_data)
- self.y_testData = x_Predict.copy()
- self.h_testData = self.Model.components_
- if Add_Func is not None and x_name != '':
- Add_Func(self.h_testData, f'{x_name}:V->NMF[H]')
- self.have_Predict = True
- return x_Predict, 'V->NMF[W]'
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- y_data = self.y_testData
- x_data = self.x_testData
- h_data = self.h_testData
- Conversion_SeparateWH(y_data, h_data, tab)
- wh_data = np.matmul(y_data, h_data)
- difference_data = x_data - wh_data
- def make_HeatMap(data, name, max_, min_):
- x = [f'数据[{i}]' for i in range(len(data))] # 主成分
- y = [f'特征[{i}]' for i in range(len(data[0]))] # 主成分
- value = [(f'数据[{i}]', f'特征[{j}]', float(data[i][j]))
- for i in range(len(data)) for j in range(len(data[i]))]
- c = (HeatMap()
- .add_xaxis(x)
- .add_yaxis(f'数据', y, value, **Label_Set) # value的第一个数值是x
- .set_global_opts(title_opts=opts.TitleOpts(title='原始数据热力图'), **global_Leg,
- yaxis_opts=opts.AxisOpts(
- is_scale=True, type_='category'), # 'category'
- xaxis_opts=opts.AxisOpts(is_scale=True, type_='category'),
- visualmap_opts=opts.VisualMapOpts(is_show=True, max_=max_,
- min_=min_,
- pos_right='3%')) # 显示
- )
- tab.add(c, name)
- max_ = max(int(x_data.max()), int(wh_data.max()),
- int(difference_data.max())) + 1
- min_ = min(int(x_data.min()), int(wh_data.min()),
- int(difference_data.min()))
- make_HeatMap(x_data, '原始数据热力图', max_, min_)
- make_HeatMap(wh_data, 'W * H数据热力图', max_, min_)
- make_HeatMap(difference_data, '数据差热力图', max_, min_)
- desTo_CSV(Dic, '权重矩阵', y_data)
- desTo_CSV(Dic, '系数矩阵', h_data)
- desTo_CSV(Dic, '系数*权重矩阵', wh_data)
- save = Dic + r'/非负矩阵分解.HTML'
- tab.render(save) # 生成HTML
- return save,
- class TSNE_Model(Unsupervised):
- def __init__(self, args_use, model, *args, **kwargs):
- super(TSNE_Model, self).__init__(*args, **kwargs)
- self.Model = TSNE(n_components=args_use['n_components'])
- self.n_components = args_use['n_components']
- self.k = {'n_components': args_use['n_components']}
- self.Model_Name = 't-SNE'
- def Fit(self, *args, **kwargs):
- self.have_Fit = True
- return 'None', 'None'
- def Predict(self, x_data, *args, **kwargs):
- self.x_testData = x_data.copy()
- x_Predict = self.Model.fit_transform(x_data)
- self.y_testData = x_Predict.copy()
- self.have_Predict = True
- return x_Predict, 'SNE'
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- y_data = self.y_testData
- Conversion_Separate_Format(y_data, tab)
- save = Dic + r'/T-SNE.HTML'
- tab.render(save) # 生成HTML
- return save,
- class MLP_Model(Study_MachineBase): # 神经网络(多层感知机),有监督学习
- def __init__(self, args_use, model, *args, **kwargs):
- super(MLP_Model, self).__init__(*args, **kwargs)
- Model = {'MLP': MLPRegressor, 'MLP_class': MLPClassifier}[model]
- self.Model = Model(
- hidden_layer_sizes=args_use['hidden_size'],
- activation=args_use['activation'],
- solver=args_use['solver'],
- alpha=args_use['alpha'],
- max_iter=args_use['max_iter'])
- # 记录这两个是为了克隆
- self.hidden_layer_sizes = args_use['hidden_size']
- self.activation = args_use['activation']
- self.max_iter = args_use['max_iter']
- self.solver = args_use['solver']
- self.alpha = args_use['alpha']
- self.k = {
- 'hidden_layer_sizes': args_use['hidden_size'],
- 'activation': args_use['activation'],
- 'max_iter': args_use['max_iter'],
- 'solver': args_use['solver'],
- 'alpha': args_use['alpha']}
- self.Model_Name = model
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- x_data = self.x_testData
- y_data = self.y_testData
- coefs = self.Model.coefs_
- class_ = self.Model.classes_
- n_layers_ = self.Model.n_layers_
- def make_HeatMap(data, name):
- x = [f'特征(节点)[{i}]' for i in range(len(data))]
- y = [f'节点[{i}]' for i in range(len(data[0]))]
- value = [(f'特征(节点)[{i}]', f'节点[{j}]', float(data[i][j]))
- for i in range(len(data)) for j in range(len(data[i]))]
- c = (HeatMap()
- .add_xaxis(x)
- .add_yaxis(f'数据', y, value, **Label_Set) # value的第一个数值是x
- .set_global_opts(title_opts=opts.TitleOpts(title=name), **global_Leg,
- yaxis_opts=opts.AxisOpts(
- is_scale=True, type_='category'), # 'category'
- xaxis_opts=opts.AxisOpts(is_scale=True, type_='category'),
- visualmap_opts=opts.VisualMapOpts(is_show=True, max_=float(data.max()),
- min_=float(data.min()),
- pos_right='3%')) # 显示
- )
- tab.add(c, name)
- tab.add(make_Tab(x, data.transpose.tolist()), f'{name}:表格')
- desTo_CSV(Dic, f'{name}:表格', data.transpose.tolist(), x, y)
- get, x_means, x_range, Type = regress_visualization(x_data, y_data)
- for i in range(len(get)):
- tab.add(get[i], f'{i}训练数据散点图')
- get = Prediction_boundary(x_range, x_means, self.Predict, Type)
- for i in range(len(get)):
- tab.add(get[i], f'{i}预测热力图')
- heard = ['神经网络层数']
- data = [n_layers_]
- for i in range(len(coefs)):
- make_HeatMap(coefs[i], f'{i}层权重矩阵')
- heard.append(f'第{i}层节点数')
- data.append(len(coefs[i][0]))
- if self.Model_Name == 'MLP_class':
- heard += [f'[{i}]类型' for i in range(len(class_))]
- data += class_.tolist()
- tab.add(make_Tab(heard, [data]), '数据表')
- save = Dic + r'/多层感知机.HTML'
- tab.render(save) # 生成HTML
- return save,
- class kmeans_Model(UnsupervisedModel):
- def __init__(self, args_use, model, *args, **kwargs):
- super(kmeans_Model, self).__init__(*args, **kwargs)
- self.Model = KMeans(n_clusters=args_use['n_clusters'])
- self.class_ = []
- self.n_clusters = args_use['n_clusters']
- self.k = {'n_clusters': args_use['n_clusters']}
- self.Model_Name = 'k-means'
- def Fit(self, x_data, *args, **kwargs):
- re = super().Fit(x_data, *args, **kwargs)
- self.class_ = list(set(self.Model.labels_.tolist()))
- self.have_Fit = True
- return re
- def Predict(self, x_data, *args, **kwargs):
- self.x_testData = x_data.copy()
- y_Predict = self.Model.predict(x_data)
- self.y_testData = y_Predict.copy()
- self.have_Predict = True
- return y_Predict, 'k-means'
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- y = self.y_testData
- x_data = self.x_testData
- class_ = self.class_
- center = self.Model.cluster_centers_
- class_heard = [f'簇[{i}]' for i in range(len(class_))]
- Func = Training_visualization_More if More_Global else Training_visualization_Center
- get, x_means, x_range, Type = Func(x_data, class_, y, center)
- for i in range(len(get)):
- tab.add(get[i], f'{i}数据散点图')
- get = Decision_boundary(x_range, x_means, self.Predict, class_, Type)
- for i in range(len(get)):
- tab.add(get[i], f'{i}预测热力图')
- heard = class_heard + [f'普适预测第{i}特征' for i in range(len(x_means))]
- data = class_ + [f'{i}' for i in x_means]
- c = Table().add(headers=heard, rows=[data])
- tab.add(c, '数据表')
- desTo_CSV(Dic, '预测表', [[f'{i}' for i in x_means]], [
- f'普适预测第{i}特征' for i in range(len(x_means))])
- save = Dic + r'/k-means聚类.HTML'
- tab.render(save) # 生成HTML
- return save,
- class Agglomerative_Model(UnsupervisedModel):
- def __init__(self, args_use, model, *args, **kwargs):
- super(Agglomerative_Model, self).__init__(*args, **kwargs)
- self.Model = AgglomerativeClustering(
- n_clusters=args_use['n_clusters']) # 默认为2,不同于k-means
- self.class_ = []
- self.n_clusters = args_use['n_clusters']
- self.k = {'n_clusters': args_use['n_clusters']}
- self.Model_Name = 'Agglomerative'
- def Fit(self, x_data, *args, **kwargs):
- re = super().Fit(x_data, *args, **kwargs)
- self.class_ = list(set(self.Model.labels_.tolist()))
- self.have_Fit = True
- return re
- def Predict(self, x_data, *args, **kwargs):
- self.x_testData = x_data.copy()
- y_Predict = self.Model.fit_predict(x_data)
- self.y_trainData = y_Predict.copy()
- self.have_Predict = True
- return y_Predict, 'Agglomerative'
- def Des(self, Dic, *args, **kwargs):
- tab = Tab()
- y = self.y_testData
- x_data = self.x_testData
- class_ = self.class_
- class_heard = [f'簇[{i}]' for i in range(len(class_))]
- Func = Training_visualization_More_NoCenter if More_Global else Training_visualization
- get, x_means, x_range, Type = Func(x_data, class_, y)
- for i in range(len(get)):
- tab.add(get[i], f'{i}训练数据散点图')
- get = Decision_boundary(x_range, x_means, self.Predict, class_, Type)
- for i in range(len(get)):
- tab.add(get[i], f'{i}预测热力图')
- linkage_array = ward(self.x_trainData) # self.y_trainData是结果
- dendrogram(linkage_array)
- plt.savefig(Dic + r'/Cluster_graph.png')
- image = Image()
- image.add(
- src=Dic + r'/Cluster_graph.png',
- ).set_global_opts(
- title_opts=opts.ComponentTitleOpts(title="聚类树状图")
- )
- tab.add(image, '聚类树状图')
- heard = class_heard + [f'普适预测第{i}特征' for i in range(len(x_means))]
- data = class_ + [f'{i}' for i in x_means]
- c = Table().add(headers=heard, rows=[data])
- tab.add(c, '数据表')
- desTo_CSV(Dic, '预测表', [[f'{i}' for i in x_means]], [
- f'普适预测第{i}特征' for i in range(len(x_means))])
- save = Dic + r'/层次聚类.HTML'
- tab.render(save) # 生成HTML
- return save,
- class DBSCAN_Model(UnsupervisedModel):
- def __init__(self, args_use, model, *args, **kwargs):
- super(DBSCAN_Model, self).__init__(*args, **kwargs)
- self.Model = DBSCAN(
- eps=args_use['eps'],
- min_samples=args_use['min_samples'])
- # eps是距离(0.5),min_samples(5)是簇与噪音分界线(每个簇最小元素数)
- # min_samples
- self.eps = args_use['eps']
- self.min_samples = args_use['min_samples']
- self.k = {
- 'min_samples': args_use['min_samples'],
- 'eps': args_use['eps']}
- self.class_ = []
- self.Model_Name = 'DBSCAN'
- def Fit(self, x_data, *args, **kwargs):
- re = super().Fit(x_data, *args, **kwargs)
- self.class_ = list(set(self.Model.labels_.tolist()))
- self.have_Fit = True
- return re
- def Predict(self, x_data, *args, **kwargs):
- self.x_testData = x_data.copy()
- y_Predict = self.Model.fit_predict(x_data)
- self.y_testData = y_Predict.copy()
- self.have_Predict = True
- return y_Predict, 'DBSCAN'
- def Des(self, Dic, *args, **kwargs):
- # DBSCAN没有预测的必要
- tab = Tab()
- y = self.y_testData.copy()
- x_data = self.x_testData.copy()
- class_ = self.class_
- class_heard = [f'簇[{i}]' for i in range(len(class_))]
- Func = Training_visualization_More_NoCenter if More_Global else Training_visualization
- get, x_means, x_range, Type = Func(x_data, class_, y)
- for i in range(len(get)):
- tab.add(get[i], f'{i}训练数据散点图')
- heard = class_heard + [f'普适预测第{i}特征' for i in range(len(x_means))]
- data = class_ + [f'{i}' for i in x_means]
- c = Table().add(headers=heard, rows=[data])
- tab.add(c, '数据表')
- desTo_CSV(Dic, '预测表', [[f'{i}' for i in x_means]], [
- f'普适预测第{i}特征' for i in range(len(x_means))])
- save = Dic + r'/密度聚类.HTML'
- tab.render(save) # 生成HTML
- return save,
- class Fast_Fourier(Study_MachineBase): # 快速傅里叶变换
- def __init__(self, args_use, model, *args, **kwargs):
- super(Fast_Fourier, self).__init__(*args, **kwargs)
- self.Model = None
- self.Fourier = None # fft复数
- self.Frequency = None # 频率range
- self.angular_Frequency = None # 角频率range
- self.Phase = None # 相位range
- self.Breadth = None # 震幅range
- self.N = None # 样本数
- def Fit(self, y_data, *args, **kwargs):
- y_data = y_data.ravel() # 扯平为一维数组
- try:
- if self.y_trainData is None:
- raise Exception
- self.y_trainData = np.hstack(y_data, self.x_trainData)
- except BaseException:
- self.y_trainData = y_data.copy()
- Fourier = fft(y_data)
- self.N = len(y_data)
- self.Frequency = np.linspace(0, 1, self.N) # 频率N_range
- self.angular_Frequency = self.Frequency / (np.pi * 2) # 角频率w
- self.Phase = np.angle(Fourier)
- self.Breadth = np.abs(Fourier)
- self.Fourier = Fourier
- self.have_Fit = True
- return 'None', 'None'
- def Predict(self, x_data, *args, **kwargs):
- return np.array([]), ''
- def Des(self, Dic, *args, **kwargs):
- # DBSCAN没有预测的必要
- tab = Tab()
- y = self.y_trainData.copy()
- N = self.N
- Phase = self.Phase # 相位range
- Breadth = self.Breadth # 震幅range
- normalization_Breadth = Breadth / N
- def line(name, value, s=slice(0, None)) -> line:
- c = (
- line() .add_xaxis(
- self.Frequency[s].tolist()) .add_yaxis(
- '',
- value,
- **Label_Set,
- symbol='none' if self.N >= 500 else None) .set_global_opts(
- title_opts=opts.TitleOpts(
- title=name),
- **global_Leg,
- xaxis_opts=opts.AxisOpts(
- type_='value'),
- yaxis_opts=opts.AxisOpts(
- type_='value')))
- return c
- tab.add(line('原始数据', y.tolist()), '原始数据')
- tab.add(line('双边振幅谱', Breadth.tolist()), '双边振幅谱')
- tab.add(
- line(
- '双边振幅谱(归一化)',
- normalization_Breadth.tolist()),
- '双边振幅谱(归一化)')
- tab.add(
- line('单边相位谱', Breadth[:int(N / 2)].tolist(), slice(0, int(N / 2))), '单边相位谱')
- tab.add(line('单边相位谱(归一化)', normalization_Breadth[:int(
- N / 2)].tolist(), slice(0, int(N / 2))), '单边相位谱(归一化)')
- tab.add(line('双边相位谱', Phase.tolist()), '双边相位谱')
- tab.add(
- line('单边相位谱', Phase[:int(N / 2)].tolist(), slice(0, int(N / 2))), '单边相位谱')
- tab.add(make_Tab(self.Frequency.tolist(), [Breadth.tolist()]), '双边振幅谱')
- tab.add(make_Tab(self.Frequency.tolist(), [Phase.tolist()]), '双边相位谱')
- tab.add(
- make_Tab(
- self.Frequency.tolist(), [
- self.Fourier.tolist()]), '快速傅里叶变换')
- save = Dic + r'/快速傅里叶.HTML'
- tab.render(save) # 生成HTML
- return save,
- class Reverse_Fast_Fourier(Study_MachineBase): # 快速傅里叶变换
- def __init__(self, args_use, model, *args, **kwargs):
- super(Reverse_Fast_Fourier, self).__init__(*args, **kwargs)
- self.Model = None
- self.N = None
- self.y_testData_real = None
- self.Phase = None
- self.Breadth = None
- def Fit(self, y_data, *args, **kwargs):
- return 'None', 'None'
- def Predict(self, x_data, x_name='', Add_Func=None, *args, **kwargs):
- self.x_testData = x_data.ravel().astype(np.complex_)
- Fourier = ifft(self.x_testData)
- self.y_testData = Fourier.copy()
- self.y_testData_real = np.real(Fourier)
- self.N = len(self.y_testData_real)
- self.Phase = np.angle(self.x_testData)
- self.Breadth = np.abs(self.x_testData)
- Add_Func(self.y_testData_real.copy(), f'{x_name}:逆向快速傅里叶变换[实数]')
- return Fourier, '逆向快速傅里叶变换'
- def Des(self, Dic, *args, **kwargs):
- # DBSCAN没有预测的必要
- tab = Tab()
- y = self.y_testData_real.copy()
- y_data = self.y_testData.copy()
- N = self.N
- range_N = np.linspace(0, 1, N).tolist()
- Phase = self.Phase # 相位range
- Breadth = self.Breadth # 震幅range
- def line(name, value, s=slice(0, None)) -> line:
- c = (
- line() .add_xaxis(
- range_N[s]) .add_yaxis(
- '',
- value,
- **Label_Set,
- symbol='none' if N >= 500 else None) .set_global_opts(
- title_opts=opts.TitleOpts(
- title=name),
- **global_Leg,
- xaxis_opts=opts.AxisOpts(
- type_='value'),
- yaxis_opts=opts.AxisOpts(
- type_='value')))
- return c
- tab.add(line('逆向傅里叶变换', y.tolist()), '逆向傅里叶变换[实数]')
- tab.add(make_Tab(range_N, [y_data.tolist()]), '逆向傅里叶变换数据')
- tab.add(make_Tab(range_N, [y.tolist()]), '逆向傅里叶变换数据[实数]')
- tab.add(line('双边振幅谱', Breadth.tolist()), '双边振幅谱')
- tab.add(
- line('单边相位谱', Breadth[:int(N / 2)].tolist(), slice(0, int(N / 2))), '单边相位谱')
- tab.add(line('双边相位谱', Phase.tolist()), '双边相位谱')
- tab.add(
- line('单边相位谱', Phase[:int(N / 2)].tolist(), slice(0, int(N / 2))), '单边相位谱')
- save = Dic + r'/快速傅里叶.HTML'
- tab.render(save) # 生成HTML
- return save,
- class Reverse_Fast_Fourier_TwoNumpy(Reverse_Fast_Fourier): # 2快速傅里叶变换
- def Fit(
- self,
- x_data,
- y_data=None,
- x_name='',
- Add_Func=None,
- *args,
- **kwargs):
- r = np.multiply(np.cos(x_data), y_data)
- j = np.multiply(np.sin(x_data), y_data) * 1j
- super(
- Reverse_Fast_Fourier_TwoNumpy,
- self).Predict(
- r + j,
- x_name=x_name,
- Add_Func=Add_Func,
- *args,
- **kwargs)
- return 'None', 'None'
- class Curve_fitting(Study_MachineBase): # 曲线拟合
- def __init__(self, Name, str_, model, *args, **kwargs):
- super(Curve_fitting, self).__init__(*args, **kwargs)
- def ndimDown(data: np.ndarray):
- if data.ndim == 1:
- return data
- new_data = []
- for i in data:
- new_data.append(np.sum(i))
- return np.array(new_data)
- NAME = {'np': np, 'Func': model, 'ndimDown': ndimDown}
- DEF = f'''
- def FUNC({",".join(model.__code__.co_varnames)}):
- answer = Func({",".join(model.__code__.co_varnames)})
- return ndimDown(answer)
- '''
- exec(DEF, NAME)
- self.Func = NAME['FUNC']
- self.Fit_data = None
- self.Name = Name
- self.Func_Str = str_
- def Fit(self, x_data: np.ndarray, y_data: np.ndarray, *args, **kwargs):
- y_data = y_data.ravel()
- x_data = x_data.astype(np.float64)
- try:
- if self.x_trainData is None:
- raise Exception
- self.x_trainData = np.vstack(x_data, self.x_trainData)
- self.y_trainData = np.vstack(y_data, self.y_trainData)
- except BaseException:
- self.x_trainData = x_data.copy()
- self.y_trainData = y_data.copy()
- self.Fit_data = optimize.curve_fit(
- self.Func, self.x_trainData, self.y_trainData)
- self.Model = self.Fit_data[0].copy()
- return 'None', 'None'
- def Predict(self, x_data, *args, **kwargs):
- self.x_testData = x_data.copy()
- Predict = self.Func(x_data, *self.Model)
- y_Predict = []
- for i in Predict:
- y_Predict.append(np.sum(i))
- y_Predict = np.array(y_Predict)
- self.y_testData = y_Predict.copy()
- self.have_Predict = True
- return y_Predict, self.Name
- def Des(self, Dic, *args, **kwargs):
- # DBSCAN没有预测的必要
- tab = Tab()
- y = self.y_testData.copy()
- x_data = self.x_testData.copy()
- get, x_means, x_range, Type = regress_visualization(x_data, y)
- for i in range(len(get)):
- tab.add(get[i], f'{i}预测类型图')
- get = Prediction_boundary(x_range, x_means, self.Predict, Type)
- for i in range(len(get)):
- tab.add(get[i], f'{i}预测热力图')
- tab.add(make_Tab([f'普适预测第{i}特征' for i in range(len(x_means))], [
- [f'{i}' for i in x_means]]), '普适预测特征数据')
- tab.add(make_Tab([f'参数[{i}]' for i in range(len(self.Model))], [
- [f'{i}' for i in self.Model]]), '拟合参数')
- save = Dic + r'/曲线拟合.HTML'
- tab.render(save) # 生成HTML
- return save,
- class Machine_Learner(Learner): # 数据处理者
- def __init__(self, *args, **kwargs):
- super().__init__(*args, **kwargs)
- self.Learner = {} # 记录机器
- self.Learn_Dic = {
- 'Line': Line_Model,
- 'Ridge': Line_Model,
- 'Lasso': Line_Model,
- 'LogisticRegression': LogisticRegression_Model,
- 'Knn_class': Knn_Model,
- 'Knn': Knn_Model,
- 'Tree_class': Tree_Model,
- 'Tree': Tree_Model,
- 'Forest': Forest_Model,
- 'Forest_class': Forest_Model,
- 'GradientTree_class': GradientTree_Model,
- 'GradientTree': GradientTree_Model,
- 'Variance': Variance_Model,
- 'SelectKBest': SelectKBest_Model,
- 'Z-Score': Standardization_Model,
- 'MinMaxScaler': MinMaxScaler_Model,
- 'LogScaler': LogScaler_Model,
- 'atanScaler': atanScaler_Model,
- 'decimalScaler': decimalScaler_Model,
- 'sigmodScaler': sigmodScaler_Model,
- 'Mapzoom': Mapzoom_Model,
- 'Fuzzy_quantization': Fuzzy_quantization_Model,
- 'Regularization': Regularization_Model,
- 'Binarizer': Binarizer_Model,
- 'Discretization': Discretization_Model,
- 'Label': Label_Model,
- 'OneHotEncoder': OneHotEncoder_Model,
- 'Missed': Missed_Model,
- 'PCA': PCA_Model,
- 'RPCA': RPCA_Model,
- 'KPCA': KPCA_Model,
- 'LDA': LDA_Model,
- 'SVC': SVC_Model,
- 'SVR': SVR_Model,
- 'MLP': MLP_Model,
- 'MLP_class': MLP_Model,
- 'NMF': NMF_Model,
- 't-SNE': TSNE_Model,
- 'k-means': kmeans_Model,
- 'Agglomerative': Agglomerative_Model,
- 'DBSCAN': DBSCAN_Model,
- 'ClassBar': Class_To_Bar,
- 'FeatureScatter': Near_feature_scatter,
- 'FeatureScatterClass': Near_feature_scatter_class,
- 'FeatureScatter_all': Near_feature_scatter_More,
- 'FeatureScatterClass_all': Near_feature_scatter_class_More,
- 'HeatMap': Numpy_To_HeatMap,
- 'FeatureY-X': Feature_scatter_YX,
- 'ClusterTree': Cluster_Tree,
- 'MatrixScatter': MatrixScatter,
- 'Correlation': CORR,
- 'Statistics': Des,
- 'Fast_Fourier': Fast_Fourier,
- 'Reverse_Fast_Fourier': Reverse_Fast_Fourier,
- '[2]Reverse_Fast_Fourier': Reverse_Fast_Fourier_TwoNumpy,
- }
- self.Learner_Type = {} # 记录机器的类型
- def p_Args(self, Text, Type): # 解析参数
- args = {}
- args_use = {}
- # 输入数据
- exec(Text, args)
- # 处理数据
- if Type in ('MLP', 'MLP_class'):
- args_use['alpha'] = float(args.get('alpha', 0.0001)) # MLP正则化用
- else:
- args_use['alpha'] = float(args.get('alpha', 1.0)) # L1和L2正则化用
- args_use['C'] = float(args.get('C', 1.0)) # L1和L2正则化用
- if Type in ('MLP', 'MLP_class'):
- args_use['max_iter'] = int(args.get('max_iter', 200)) # L1和L2正则化用
- else:
- args_use['max_iter'] = int(args.get('max_iter', 1000)) # L1和L2正则化用
- args_use['n_neighbors'] = int(args.get('K_knn', 5)) # knn邻居数 (命名不同)
- args_use['p'] = int(args.get('p', 2)) # 距离计算方式
- args_use['nDim_2'] = bool(args.get('nDim_2', True)) # 数据是否降维
- if Type in ('Tree', 'Forest', 'GradientTree'):
- args_use['criterion'] = 'mse' if bool(
- args.get('is_MSE', True)) else 'mae' # 是否使用基尼不纯度
- else:
- args_use['criterion'] = 'gini' if bool(
- args.get('is_Gini', True)) else 'entropy' # 是否使用基尼不纯度
- args_use['splitter'] = 'random' if bool(
- args.get('is_random', False)) else 'best' # 决策树节点是否随机选用最优
- args_use['max_features'] = args.get('max_features', None) # 选用最多特征数
- args_use['max_depth'] = args.get('max_depth', None) # 最大深度
- args_use['min_samples_split'] = int(
- args.get('min_samples_split', 2)) # 是否继续划分(容易造成过拟合)
- args_use['P'] = float(args.get('min_samples_split', 0.8))
- args_use['k'] = args.get('k', 1)
- args_use['score_func'] = (
- {
- 'chi2': chi2,
- 'f_classif': f_classif,
- 'mutual_info_classif': mutual_info_classif,
- 'f_regression': f_regression,
- 'mutual_info_regression': mutual_info_regression}. get(
- args.get(
- 'score_func',
- 'f_classif'),
- f_classif))
- args_use['feature_range'] = tuple(args.get('feature_range', (0, 1)))
- args_use['norm'] = args.get('norm', 'l2') # 正则化的方式L1或者L2
- args_use['threshold'] = float(args.get('threshold', 0.0)) # 二值化特征
- args_use['split_range'] = list(args.get('split_range', [0])) # 二值化特征
- args_use['ndim_up'] = bool(args.get('ndim_up', False))
- args_use['miss_value'] = args.get('miss_value', np.nan)
- args_use['fill_method'] = args.get('fill_method', 'mean')
- args_use['fill_value'] = args.get('fill_value', None)
- args_use['n_components'] = args.get('n_components', 1)
- args_use['kernel'] = args.get(
- 'kernel', 'rbf' if Type in (
- 'SVR', 'SVC') else 'linear')
- args_use['n_Tree'] = args.get('n_Tree', 100)
- args_use['gamma'] = args.get('gamma', 1)
- args_use['hidden_size'] = tuple(args.get('hidden_size', (100,)))
- args_use['activation'] = str(args.get('activation', 'relu'))
- args_use['solver'] = str(args.get('solver', 'adam'))
- if Type in ('k-means',):
- args_use['n_clusters'] = int(args.get('n_clusters', 8))
- else:
- args_use['n_clusters'] = int(args.get('n_clusters', 2))
- args_use['eps'] = float(args.get('n_clusters', 0.5))
- args_use['min_samples'] = int(args.get('n_clusters', 5))
- args_use['white_PCA'] = bool(args.get('white_PCA', False))
- return args_use
- def Add_Learner(self, Learner, Text=''):
- get = self.Learn_Dic[Learner]
- name = f'Le[{len(self.Learner)}]{Learner}'
- # 参数调节
- args_use = self.p_Args(Text, Learner)
- # 生成学习器
- self.Learner[name] = get(model=Learner, args_use=args_use)
- self.Learner_Type[name] = Learner
- def Add_Curve_Fitting(self, Learner_text, Text=''):
- NAME = {}
- exec(Learner_text, NAME)
- name = f'Le[{len(self.Learner)}]{NAME.get("name","SELF")}'
- func = NAME.get('f', lambda x, k, b: k * x + b)
- self.Learner[name] = Curve_fitting(name, Learner_text, func)
- self.Learner_Type[name] = 'Curve_fitting'
- def Add_SelectFrom_Model(self, Learner, Text=''): # Learner代表选中的学习器
- model = self.get_Learner(Learner)
- name = f'Le[{len(self.Learner)}]SelectFrom_Model:{Learner}'
- # 参数调节
- args_use = self.p_Args(Text, 'SelectFrom_Model')
- # 生成学习器
- self.Learner[name] = SelectFrom_Model(
- Learner=model, args_use=args_use, Dic=self.Learn_Dic)
- self.Learner_Type[name] = 'SelectFrom_Model'
- def Add_Predictive_HeatMap(self, Learner, Text=''): # Learner代表选中的学习器
- model = self.get_Learner(Learner)
- name = f'Le[{len(self.Learner)}]Predictive_HeatMap:{Learner}'
- # 生成学习器
- args_use = self.p_Args(Text, 'Predictive_HeatMap')
- self.Learner[name] = Predictive_HeatMap(
- Learner=model, args_use=args_use)
- self.Learner_Type[name] = 'Predictive_HeatMap'
- def Add_Predictive_HeatMap_More(self, Learner, Text=''): # Learner代表选中的学习器
- model = self.get_Learner(Learner)
- name = f'Le[{len(self.Learner)}]Predictive_HeatMap_More:{Learner}'
- # 生成学习器
- args_use = self.p_Args(Text, 'Predictive_HeatMap_More')
- self.Learner[name] = Predictive_HeatMap_More(
- Learner=model, args_use=args_use)
- self.Learner_Type[name] = 'Predictive_HeatMap_More'
- def Add_View_data(self, Learner, Text=''): # Learner代表选中的学习器
- model = self.get_Learner(Learner)
- name = f'Le[{len(self.Learner)}]View_data:{Learner}'
- # 生成学习器
- args_use = self.p_Args(Text, 'View_data')
- self.Learner[name] = View_data(Learner=model, args_use=args_use)
- self.Learner_Type[name] = 'View_data'
- def Return_Learner(self):
- return self.Learner.copy()
- def get_Learner(self, name):
- return self.Learner[name]
- def get_Learner_Type(self, name):
- return self.Learner_Type[name]
- def Fit(self, x_name, y_name, Learner, split=0.3, *args, **kwargs):
- x_data = self.get_Sheet(x_name)
- y_data = self.get_Sheet(y_name)
- model = self.get_Learner(Learner)
- return model.Fit(
- x_data,
- y_data,
- split=split,
- x_name=x_name,
- Add_Func=self.Add_Form)
- def Predict(self, x_name, Learner, Text='', **kwargs):
- x_data = self.get_Sheet(x_name)
- model = self.get_Learner(Learner)
- y_data, name = model.predict(
- x_data, x_name=x_name, Add_Func=self.Add_Form)
- self.Add_Form(y_data, f'{x_name}:{name}')
- return y_data
- def Score(self, name_x, name_y, Learner): # Score_Only表示仅评分 Fit_Simp 是普遍类操作
- model = self.get_Learner(Learner)
- x = self.get_Sheet(name_x)
- y = self.get_Sheet(name_y)
- return model.Score(x, y)
- def Show_Score(self, Learner, Dic, name_x, name_y, Func=0): # 显示参数
- x = self.get_Sheet(name_x)
- y = self.get_Sheet(name_y)
- if NEW_Global:
- dic = Dic + f'/{Learner}分类评分[CoTan]'
- new_dic = dic
- a = 0
- while exists(new_dic): # 直到他不存在 —— False
- new_dic = dic + f'[{a}]'
- a += 1
- mkdir(new_dic)
- else:
- new_dic = Dic
- model = self.get_Learner(Learner)
- # 打包
- func = [
- model.Class_Score,
- model.Regression_Score,
- model.Clusters_Score][Func]
- save = func(new_dic, x, y)[0]
- if TAR_Global:
- make_targz(f'{new_dic}.tar.gz', new_dic)
- return save, new_dic
- def Show_Args(self, Learner, Dic): # 显示参数
- if NEW_Global:
- dic = Dic + f'/{Learner}数据[CoTan]'
- new_dic = dic
- a = 0
- while exists(new_dic): # 直到他不存在 —— False
- new_dic = dic + f'[{a}]'
- a += 1
- mkdir(new_dic)
- else:
- new_dic = Dic
- model = self.get_Learner(Learner)
- if (not(model.Model is None) or not(model.Model is list)) and CLF_Global:
- joblib.dump(model.Model, new_dic + '/MODEL.model') # 保存模型
- # pickle.dump(model,new_dic + f'/{Learner}.pkl')#保存学习器
- # 打包
- save = model.Des(new_dic)[0]
- if TAR_Global:
- make_targz(f'{new_dic}.tar.gz', new_dic)
- return save, new_dic
- def Del_Leaner(self, Leaner):
- del self.Learner[Leaner]
- del self.Learner_Type[Leaner]
- def make_targz(output_filename, source_dir):
- with tarfile.open(output_filename, "w:gz") as tar:
- tar.add(source_dir, arcname=basename(source_dir))
- return output_filename
- def set_Global(
- More=More_Global,
- All=All_Global,
- CSV=CSV_Global,
- CLF=CLF_Global,
- TAR=TAR_Global,
- NEW=NEW_Global):
- global More_Global, All_Global, CSV_Global, CLF_Global, TAR_Global, NEW_Global
- More_Global = More # 是否使用全部特征绘图
- All_Global = All # 是否导出charts
- CSV_Global = CSV # 是否导出CSV
- CLF_Global = CLF # 是否导出模型
- TAR_Global = TAR # 是否打包tar
- NEW_Global = NEW # 是否新建目录
|