데이터 활용과 투자이야기

import FinanceDataReader as fdr
df = fdr.DataReader('AAPL', '2024-1-20')
df

# 당일 종가 - 하루전 종가 구하기!
df['Close_diff'] = df['Close'].diff()
df

df['Close_diff_2'] = df['Close'].diff(2)
df

df['Close_diff_-1'] = df['Close'].diff(-1)
df

# diff 함수의 인자값 디폴트값은 (1, axis=0) 의미 
# ( 한칸전을 계산, raw 단위로 계산 ) 만약 axis=1이면 column 단위로 계산함
df['Close'].diff(axis=1)

df['Close']는 시리즈 타입 데이터라서 열이 1개이므로 axis=1로 연산을 위해

옆 열을 보니 없어서 에러가 나는 것임

import FinanceDataReader as fdr
df = fdr.DataReader('AAPL', '2024-1-20')
df

# pct_change(2) # 2는 이틀전의 값과 등락률을 구하는 것이다.
df['pct_change_2'] = df['Close'].pct_change(2)
df

# 위에 df['pct_change'] = df['Close'].pct_change() 처럼 
# pct_change가 없다면 아래와 같이 복잡하게 계산해야 함
df['aaa']=(df['Close']-df['Close'].shift(1))/df['Close'].shift(1)
df

import FinanceDataReader as fdr
df = fdr.DataReader('AAPL', '2024-1-20')
df

df['Close_lag1'] = df['Close'].shift()
df

df['Close_lag3'] = df['Close'].shift(3)
df

df['Close_lag_-1'] = df['Close'].shift(-1)
df

import FinanceDataReader as fdr
df = fdr.DataReader('005930', '2024-1-1')
df

import pandas as pd
import numpy as np

s1 = pd.Series([np.nan, 2,3,4,5])
s2 = pd.Series([1,2,np.nan,np.nan,5])
s3 = pd.Series([1,2,3,4,5])

df = pd.DataFrame({'s1':s1, 's2':s2 , 's3':s3 })
df

df['s1'].isna()
df.isna()

df.isna().sum()
df.isnull()
df.isnull().sum()

결측치 종류 3가지
np.nan / np.inf / -np.inf (inf:무한대)

# df에 있는 값이 아래 상태에 해당하는지 전수 확인
df
df.isin([np.nan, np.inf, -np.inf])
df.isin([np.nan, np.inf, -np.inf]).sum()

df.dropna() nan값이 있는 행기준으로 제거

(디폴트 axis = row 또는 0) df.dropna(axis='columns') 또는 axis = 1 는 열을 기준으로 제거

df.dropna()
df.dropna(axis=0)
df.dropna(axis='columns')
df.dropna(axis=1)

# df.fillna(9) na값을 모두 9로 바꾼다.
df.fillna(9)

# df.fillna(method='pad')는 한 행전에 있는 값으로 대체 
# df.fillna(method='bfill')는 한 행후에 있는 값으로 대체
df.fillna(method='pad')
df.fillna(method='bfill')

# 데이터 가져오기
import FinanceDataReader as fdr
df = fdr.DataReader('005930', '2024-1-1')
df

# df.loc

df.loc[] 는 loc [ 행, 열 ] 개념으로 다룬다.

df.loc['2024-01-08','Open']
# 출력
# np.float64(77000.0)

df.loc['2024-01-08',['Open','High','Low','Close','Volume','Change']]
# 출력
# Open      7.700000e+04
# High      7.750000e+04
# Low       7.640000e+04
# Close     7.650000e+04
# Volume    1.108872e+07
# Change   -1.305483e-03
# Name: 2024-01-08 00:00:00, dtype: float64

df.loc['2024-01-08']
# 출력
# Open      7.700000e+04
# High      7.750000e+04
# Low       7.640000e+04
# Close     7.650000e+04
# Volume    1.108872e+07
# Change   -1.305483e-03
# Name: 2024-01-08 00:00:00, dtype: float64

df.loc['2024-01-08'] 처럼 한 행을 가져오는 것처럼 한 열만 가져오려면 df.loc['Open']를 하면 에러남 반드시 첫 인자는 행이름이 들어가야 함 그래서 df.loc[:,'Open']로 표현함

df.loc[:,'Open']
df.loc[['2024-01-05','2024-01-08'],['High','Low']]

# df.iloc

df.iloc[] 는 [ 행, 열 ] 행과열을 모두 숫자 index를 개념으로 다룬다.

df.iloc[1:5,1:4] 처럼 정수 슬라이싱을 이용하면 된다.

df.iloc[1,3]
df.iloc[1:5,1:4]

https://finance.yahoo.com/quote/005930.KS/history/ 
로 제공해주는 라이브러리가 있었는데 현재는 뭔가 변경되었는지  데이터가 출력이 안나옴 (2025-01-01 기준)

그래서 주로 아래 라이브러리를 사용하기로 함

# 설치하기

pip3.12 install -U finance-datareader / -U는 -upgrade 옵션과 동일

pip3.12 install plotly ◀ import FinanceDataReader시 이게 없으면 에러남

FinanceDataReader 사용자 안내서

https://financedata.github.io/posts/finance-data-reader-users-guide.html

import FinanceDataReader as fdr
df = fdr.DataReader('005930', '2024-1-1')

# 2024-01-08에 open(시가)를 가져와 보기
df['Open']['2024-01-08']

# df[['Open','Close']] 에서 바깥 []는 데이터프레임의 인덱스로써의 [] 기호이고 
# 안쪽에 []는 여러개의 컬럼을 기술하기 위한 리스트 개념의 [] 기호이니 주의!!

df[['Open','Close']]

df[0:5]
# df[0:5]를 넣으면 위네 처럼 컬럼이 나올것 같지만 5개행이 출력됨 주의!! 
# 또 아래와 같이 df[['Open','Close']]['2024-01-08'] 이렇게 하면 에러남

df['2024-01-04':'2024-01-08']
# df[0:5]에서는 오른쪽 5는 포함이 안되어 5행을 가져왔지만 
# df['2024-01-04':'2024-01-08']에서는 오른쪽 '2024-01-08'의 값도 가져옴

# CSV 불러와서 타입 바로잡기

import pyspark.pandas as ps
from pyspark.sql import SparkSession
import pandas as pd

df = spark.read.format("csv").load("/FileStore/tables/ACE_BigTech_Top7_2X.csv",encoding='cp949', header=True).toPandas()
# df.head(30)

df['이격1'] = df['이격1'].astype(float)
df.dtypes

import pandas as pd
my_list  =  [('a',  2,  3),
('b',  5,  6),
('c',  8,  9),
('a',  2,  3),
('b',  5,  6),
('c',  8,  9)]
col_name  =  ['col1',  'col2',  'col3']
#-- pd
dp  =  pd.DataFrame(my_list,columns=col_name)
dp
#-- ps
ds = spark.createDataFrame(my_list,schema=col_name

import pyspark.sql.functions as F
#-- pd
dp['concat']  =  dp.apply(lambda  x:'%s%s'%(x['col1'],x['col2']),axis=1)
dp
#-- ps
ds.withColumn('concat',F.concat('col1','col2')).show()

# 5.3.14 GroupBy
#-- pd
dp.groupby(['col1']).agg({'col2':'min','col3':'mean'})
#-- ps
ds.groupBy(['col1']).agg({'col2': 'min', 'col3': 'avg'}).show()

# 5.3.15 피벗(Pivot)
import numpy as np
#-- pd
pd.pivot_table(dp, values='col3', index='col1', columns='col2', aggfunc=np.sum)
#-- ps
ds.groupBy(['col1']).pivot('col2').sum('col3').show()

# 5.3.16  Window
d  =  {'A':['a','b','c','d'],'B':['m','m','n','n'],'C':[1,2,3,6]}
#-- pd
dp  =  pd.DataFrame(d)
dp
#-- ps
ds  =  spark.createDataFrame(dp)
ds.show()

#-- pd
dp['rank'] = dp.groupby('B')['C'].rank('dense',ascending=False)
dp
#-- ps
from pyspark.sql.window import Window
w = Window.partitionBy('B').orderBy(ds.C.desc())
ds  =  ds.withColumn('rank',F.rank().over(w))
ds.show()

# 5.3.17 rank vs dense_rank
d  ={'Id':[1,2,3,4,5,6],
'Score':  [4.00,  4.00,  3.85,  3.65,  3.65,  3.50]}
data = pd.DataFrame(d)
dp = data.copy()
dp
ds = spark.createDataFrame(data)

#-- ps
dp['Rank_dense'] = dp['Score'].rank(method='dense',ascending=False)
dp['Rank'] = dp['Score'].rank(method='min',ascending=False)
dp
#-- ps
import  pyspark.sql.functions  as  F
from  pyspark.sql.window  import  Window
w  =  Window.orderBy(ds.Score.desc())
ds  =  ds.withColumn('Rank_spark_dense',F.dense_rank().over(w)) ds  =  ds.withColumn('Rank_spark',F.rank().over(w))
ds.show()

#-- pd --
import pandas as pd
leftp  =  pd.DataFrame({'A':  ['A0',  'A1',  'A2',  'A3'],
        'B':  ['B0',  'B1',  'B2',  'B3'],
        'C':  ['C0',  'C1',  'C2',  'C3'],
        'D':  ['D0',  'D1',  'D2',  'D3']}, index=[0,  1,  2,  3])
rightp  =  pd.DataFrame({'A':  ['A0',  'A1',  'A6',  'A7'],
        'F':  ['B4',  'B5',  'B6',  'B7'], 
        'G':  ['C4',  'C5',  'C6',  'C7'],
        'H':  ['D4',  'D5',  'D6',  'D7']}, index=[4,  5,  6,  7])
#-- ps
lefts  =  spark.createDataFrame(leftp) 
rights  =  spark.createDataFrame(rightp)

leftp.head(4)
rightsp.head(4)
lefts.show(4)
rights.show(4)

# 1.  Left Join
#-- pd
leftp.merge(rightp, on='A', how='left')
#-- ps
lefts.join(rights, on='A', how='left').orderBy('A',ascending=True).show()

# 2.  Right Join
#-- pd
leftp.merge(rightp, on='A', how='right')
#-- ps
lefts.join(rights, on='A', how='right').orderBy('A',ascending=True).show()

# 3.  Inner Join
#-- pd
leftp.merge(rightp, on='A', how='inner')
#-- ps
lefts.join(rights, on='A', how='inner').orderBy('A',ascending=True).show()

# 4.  Full Join
#-- pd
leftp.merge(rightp, on='A', how='outer')
#-- ps
lefts.join(rights, on='A', how='full').orderBy('A',ascending=True).show()

# pd (판다스)

import pandas as pd
# import pyspark.pandas as ps

my_list = [[230.1, 37.8, 69.2, 22.1], [44.5, 39.3, 45.1, 10.4],[17.2, 45.9, 69.3, 9.3],[151.5, 41.3, 58.5, 18.5]]
col_name = ['TV', 'Radio', 'Newspaper', 'Sales']

#-- pd --
drop_name = ['Newspaper', 'Sales']
dp = pd.DataFrame(my_list, columns= col_name)
dp.drop(drop_name, axis=1).head(4)

# PS (spark)

import pyspark.pandas as ps
spark = SparkSession .builder \
    .appName("Python Spark create RDD example") \
    .config("spark.some.config.option", "some-value") \
    .getOrCreate()
ds = spark.read.format("csv").load("/FileStore/tables/Advertising.csv", header=True)
#-- ps --
drop_name = ['Newspaper', 'Sales']
ds.drop(*drop_name).show(4)

5.3.10 추출(Filter)

# pd (판다스)

import pyspark.pandas as ps
from pyspark.sql import SparkSession
# 주의 !! databricks에서 pandas의 csv파일 read하려면 로컬에 파일이 있어야만 가능하기 때문에 교재명령은 불가능
# dp = pd.read_csv('Advertising.csv') 교재예시
# 그래서 편법으로 아래와 같이 사용함

import pandas as pd

df_pandas = spark.read.format("csv").load("/FileStore/tables/Advertising.csv", header=True).toPandas()
df_pandas.head(4)

df_pandas['TV'] = df_pandas['TV'].astype(float)
df_pandas['Radio'] = df_pandas['Radio'].astype(float)
df_pandas['Newspaper'] = df_pandas['Newspaper'].astype(float)
df_pandas['Sales'] = df_pandas['Sales'].astype(float)

# df_pandas.dtypes
df_pandas[df_pandas.Newspaper<20].head(4)
df_pandas[(df_pandas.Newspaper<20)&(df_pandas.TV>100)].head(4)

# PS (spark)

ds = spark.read.format("csv").load("/FileStore/tables/Advertising.csv", header=True)
ds[ds.Newspaper<20].show(4)
ds[(ds.Newspaper<20)&(ds.TV>100)].show(4)

5.3.11 새 컬럼 추가하기

#-- df_pandas
df_pandas['tv_norm'] = df_pandas.TV/sum(df_pandas.TV)
df_pandas.head(4)

#-- ps
import pyspark.sql.functions as F
ds = spark.read.format("csv").load("/FileStore/tables/Advertising.csv", header=True)
ds.withColumn('tv_norm', ds.TV/ds.groupBy().agg(F.sum("TV")).collect()[0][0]).show(4)

#-- df_pandas
df_pandas['cond'] = df_pandas.apply(lambda A: 1 if((A.TV>100)&(A.Radio<40)) else 2 if(A.Sales> 10) else 3,axis=1)
df_pandas.head(4)

#-- ps
ds.withColumn('cond', F.when((ds.TV>100)&(ds.Radio<40),1)\
    .when(ds.Sales>10, 2)\
    .otherwise(3)).show(4)

#-- df_pandas
import numpy as np
df_pandas['log_tv'] = np.log(df_pandas.TV)
df_pandas.head(4)

#-- ps
import pyspark.sql.functions as F
ds.withColumn('log_tv', F.log(ds.TV)).show(4)

#-- df_pandas
df_pandas['tv+10'] = df_pandas.TV.apply(lambda x: x+10)
df_pandas.head(4)

#-- ps
ds.withColumn('tv+10', ds.TV+10).show(4)