해당 feature에 대한 설명

.python-version

@@ -1 +1,2 @@
 airflow
+#3.8.10

README.md

@@ -12,7 +12,7 @@
   <li>upbit</li>
   <li>Binance</li>
 </ul>
-
+만약 해당 dags를 실행하고자 한다면 my_package파일을 python path에 적용 시켜 주셔야 합니다.
 <br>
 <br>
 <hr>

dags/market/extrection/coin_name.py

@@ -7,6 +7,8 @@
 from upbit_def import min_data
 
 local_tz = pendulum.timezone("Asia/Seoul")
+execution_date='{{execution_date.add(hours=9).strftime("%Y%m%d")}}/'
+data_local_route='$data_route'
 
 ####이름 정보를 가져오는 dag
 default_args = {

dags/market/extrection/minute_data.py

@@ -32,7 +32,7 @@
         bash_command="echo 'END'"
     )
 
-    exe_date = '{{execution_date.add(hours=9).strftime("%Y%m%d%")}}'
+    exe_date = '{{execution_date.add(hours=9).strftime("%Y%m%d")}}'
     exe_hour = '{{execution_date.add(hours=9).strftime("%H")}}'
     exe_min = '{{execution_date.add(hours=9).strftime(%M)}}'
     local_min = f'$HOME/tmp/data/minute/{exe_date}/{exe_hour}/{exe_min}'
@@ -66,4 +66,4 @@
         bash_command=f'{line_mesg}'
     )
 
-start >> dir_make >> calling_name >> upload_to_dfs >> error >> end
+start >> dir_make >> calling_name >> upload_to_dfs >> [error,end]

dags/market/extrection/old_min_data.py

@@ -0,0 +1,53 @@
+from airflow import DAG
+from airflow.operators.bash import BashOperator
+from airflow.operators.python import PythonOperator
+from airflow.operators.empty import EmptyOperator
+from datetime import datetime
+import pendulum
+from upbit_def import min_data,api_with_execute
+
+
+local_tz = pendulum.timezone("Asia/Seoul")
+line_mesg = "curl -X POST -H 'Authorization: Bearer 6Y3IVP0dZD9bREhqMS4pd0sZZg5QAh3N9eAcixrovns' -F 'message=this task has been error' https://notify-api.line.me/api/notify"
+s_line_mesg = "curl -X POST -H 'Authorization: Bearer 6Y3IVP0dZD9bREhqMS4pd0sZZg5QAh3N9eAcixrovns' -F 'message=this task has been completed' https://notify-api.line.me/api/notify"
+
+####DAGS
+default_args = {
+    'owner': 'merlin',
+    'depends_on_past': False,
+    'start_date': datetime(2023, 7, 5,tzinfo=local_tz),
+    'retries': 5,
+}
+
+with DAG(
+    'b4_data',
+    default_args=default_args,
+    schedule_interval='* * * * *'
+)as dag:
+    start=BashOperator(
+        task_id = 'start',
+        bash_command="echo 'start'" 
+    )
+    end = BashOperator(
+        task_id = 'END',
+        bash_command="echo 'END'"
+    )
+    data_saver=PythonOperator(
+        task_id='name_roller',
+        python_callable=api_with_execute,
+        op_args=[route,name_file]
+        )
+    fail_messager=BashOperator(
+        task_id='messager',
+        bash_command=f"{line_mesg}"
+        )
+    sussced_message=BashOperator(
+        task_id='sussced_message',
+        bash_command=f"{s_line_mesg}"
+        )
+
+
+
+    start >> data_saver >> [fail_messager,sussced_message] >> end
+
+

my_package/upbit_def.py

@@ -31,6 +31,7 @@
 #####데이터베이스 테이블 이름은 -가 사용 불가하여 _로 변경
 def min_data(savePath):
     json_name = f'{savePath}/name.json'
+
     df = pd.read_json(json_name)
 
     df.drop(columns=['korean_name','english_name'],axis=1,inplace=True)
@@ -65,4 +66,96 @@ def compare_by_df(origin,new):
 
         return print("name data didn't changed")
 
+def status_adder(name_path):
+
+    name_df = pd.read_json(name_path)
+
+    name_df['status'] = 'non_start'
+
+    name_df.set_index('status', inplace=True)
+
+    name_df.to_json('name_path')
+
+def status_checker(name_path):
+
+    name_df=pd.read_json(name_path)
+
+    for index, row in name_df.iterrows():
+
+        if row['status'] == 'non_started' or row['status'] == 'progress':
+
+            name_df.at[index,'status'] = 'progress'
+
+            return row['market']
+
+def count_documents_in_folder(folder_path):
+
+    if os.path.exists(folder_path) and os.path.isdir(folder_path):
+
+        files = os.listdir(folder_path)
+
+        document_count = len(files)
+
+        return document_count
+
+    else:
+
+        return 0
+
+
+def execute_date_checker(route_path,now_market):
+
+    num_file=count_documents_in_folder(route_path)
+
+    data_path=f"{route_path}/{now_market}/{num_file}"
+
+    data_df=pd.read_json(data_path)
+
+    sort_data_df=data_df.sort_values(by='candle_date_time_kst',ascending=False)
+
+    b4_slicing_date=sort_data_df.loc[0,'candle_date_time_kst']
+    #"candle_date_time_kst":"2023-08-18T00:59:00"
+    #{yyyy}-{MM}-{dd}T{hh}%3A{mm}%3A{ss}
+    kst_year=b4_slicing_date[:4]
+    kst_month=b4_slicing_date[5:7]
+    kst_day=b4_slicing_date[8:10]
+    kst_hour=b4_slicing_date[11:13]
+    kst_min=b4_slicing_date[14:16]
+
+
+    execute_date=f"{kst_year}-{kst_month}-{kst_day}T{kst_hour}%3A{kst_min}%3A00"
+
+    list_for_save=[execute_date,num_file]
+
+    return list_for_save
+
+
+def api_calling_upbit(market,date_info):
+
+    url = f"https://api.upbit.com/v1/candles/minutes/1?market={market}&to={date_info}%2B09%3A00&count=180"
+
+    headers = {"accept": "application/json"}
+
+    response = requests.get(url, headers=headers)
+
+    parsed_data=json_load(response)
+
+    data_for_save=pd.DataFrame(parsed_data)
+
+    return data_for_save
+
+def api_with_execute(route_path,name_path):
+
+    now_market=status_checker(name_path)
+
+    execute_date_info=execute_date_checker(route_path,now_market)
+
+    api_data=api_calling_upbit(now_market,execute_date_info[0])
+
+    api_data.to_json(f"{route_path}/{execute_date_info[1]}")
+
+
+
+
+