ABOUT ME

-

Today
-
Yesterday
-
Total
-
  • 키움 계좌 정리
    Python/파이썬과 주식, 코인 2025. 4. 16. 11:04
    반응형

    계좌에서 한 종목 발라내는 게 꽤나 유용해서...
    키움에서도 한번...

    키움의 0365 창에서 조회 후 우클릭 하면
    데이터를 다운로드할 수 있다. 
    1줄 보기 체크를 잊지 말 것.

    엑셀로 저장인데, 실제로는 CSV.
    코딩에는 CSV가 더 좋다~!

    ini 파일을 만들어주고....
    ini 파일로는 리스트나 딕셔너리 형식으로 설정값을 만들기 쉽지 않다.
    JSON을 쓰면 가능한데... 몇 개 안되서... 하나씩 주석처리 풀어서..

    ;viewer.ini
    [viewer]
    db=db.sqlite3
    ; csv=210417_220416.csv
    ; csv=220417_230416.csv
    csv=230417_240416.csv
    ; csv=240417_250415_1.csv
    ; csv=240417_250415_2.csv
    price=3385
    check_only=False

    sqlite를 사용했다. 
    파이썬 내장 라이브러리만으로 간단하게... 뚝딱...

    # Stock Transaction History Viewer
    import csv
    import sqlite3
    from os.path import exists
    from configparser import ConfigParser
    from datetime import datetime, time
    
    config = ConfigParser()
    config.read("viewer.ini")
    
    DB = config['viewer']['db']
    CSV = config['viewer']['csv']
    PRICE = float(config['viewer']['price'])
    CHECK_ONLY = config['viewer']['check_only'].lower() == "true"
    
    
    def print_args(func):
        def wrapper(*args, **kwargs):
            print(*args, **kwargs)
            return func(*args, **kwargs)
    
        return wrapper
    
    
    # @print_args
    def db_dml(*args, **kwargs):
        with sqlite3.connect(DB) as conn:
            cur = conn.cursor()
            cur.execute(*args, **kwargs)
            conn.commit()
    
    
    # @print_args
    def db_query(*args, **kwargs):
        with sqlite3.connect(DB) as conn:
            cur = conn.cursor()
            cur.execute(*args, **kwargs)
            return cur.fetchall()
    
    
    def insert_data(name, datetime_, type_, stock, money, charge, tax):
        check_sql = (
            "SELECT COUNT(*) FROM buy_sell "
            "WHERE name=? AND datetime=? AND type=? AND stock=? AND money=? AND charge=? AND tax=?"
        )
        count = db_query(check_sql, (name, datetime_, type_, stock, money, charge, tax))[0][0]
    
        if count > 0:
            print(
                f"\n이미 같은 데이터가 존재합니다\n{name} {datetime_} {type_} {stock} {money} {charge} {tax}"
            )
            choice = input("데이터를 추가하시겠습니까? (y/n) : ").strip().lower()
            if choice != "y":
                return
    
        print("[INSERT]", name, datetime_, type_, stock, money, charge, tax)
        sql = (
            "INSERT INTO buy_sell(name,datetime,type,stock,money,charge,tax) "
            "VALUES (?,?,?,?,?,?,?)"
        )
        db_dml(sql, (name, datetime_, type_, stock, money, charge, tax))
    
    
    def csv_read(file, encoding="euc-kr"):
        with open(file, "r", encoding=encoding) as f:
            yield from csv.DictReader(f)
    
    
    def str2float(num: str):
        num = num.replace(",", "").strip()
        if num == "":
            return 0
        return float(num)
    
    
    def correct_time(거래일자: str, 처리시간: str) -> str:
        datetime_str = f"{거래일자} {처리시간}"
        datetime_obj = datetime.strptime(datetime_str, "%Y/%m/%d %H:%M:%S")
        # if 0 <= datetime_obj.hour < 9:
        #     datetime_obj = datetime.combine(
        #         datetime_obj.date(),
        #         time(datetime_obj.hour + 12, datetime_obj.minute, datetime_obj.second)
        #     )
        return datetime_obj.strftime("%Y/%m/%d %H:%M:%S")
    
    
    def main():
        # init db
        if not exists(DB):
            db_dml(
                """
                CREATE TABLE "buy_sell" (
                    "name" TEXT NOT NULL,
                    "ticker" TEXT,
                    "datetime"	TEXT NOT NULL,
                    "type"	TEXT NOT NULL,
                    "stock"	REAL NOT NULL DEFAULT 0,
                    "money"	REAL NOT NULL DEFAULT 0,
                    "charge"	REAL NOT NULL DEFAULT 0,
                    "tax"	INTEGER NOT NULL DEFAULT 0,
                    "note"	TEXT
                );
                """
            )
    
        for each in csv_read(CSV):
            if not CHECK_ONLY:
                insert_data(
                    each["종목명"],
                    correct_time(each['거래일자'], each['처리시간']),
                    each["거래종류"].strip(),
                    str2float(each["거래수량"]),
                    str2float(each["거래금액"]),
                    str2float(each["수수료"]),
                    str2float(each["'거래세/농특세"]) + str2float(each["'소득세/주민세"]),
                )
    
        buy = db_query(
            "SELECT sum(stock), sum(money), sum(charge), sum(tax) "
            'FROM buy_sell '
            'WHERE type="보통매매 장내매수"'
        )
        sell = db_query(
            "SELECT sum(stock), sum(money), sum(charge), sum(tax) "
            'FROM buy_sell '
            'WHERE type="보통매매 장내매도" OR type="배당금입금" OR type="대여배당금입금"'
        )
    
        buy_stock, buy_money, buy_charge, buy_tax = (
            0 if each is None else each for each in buy[0]
        )
        print(f"[매수] 수량: {buy_stock:,.0f} 금액: {buy_money:,.0f} "
              f"수수료: {buy_charge:,.0f} 세금: {buy_tax:,.0f}")
    
        sell_stock, sell_money, sell_charge, sell_tax = (
            0 if each is None else each for each in sell[0]
        )
        print(f"[매도 및 배당] 수량: {sell_stock:,.0f} 금액: {sell_money:,.0f} "
              f"수수료: {sell_charge:,.0f} 세금: {sell_tax:,.0f}")
        roi = (
                ((buy_stock - sell_stock) * PRICE + (sell_money - sell_charge - sell_tax))
                / (buy_money + buy_charge + buy_tax) * 100
        )
        print(f"ROI: {roi:.2f} %")
    
    
    if __name__ == "__main__":
        main()
    [매수] 수량: 1,808 금액: 8,758,345 수수료: 1,160 세금: 0
    [매도 및 배당] 수량: 168 금액: 691,786 수수료: 70 세금: 20,188
    ROI: 71.04 %

    오늘의 결론 - 더 열심히 물타기... --;

    반응형
Designed by Tistory.