schwab.clj

(ns u.schwab
  (:require
   [cheshire.core :as json]
   [hato.websocket :as ws]
   [clj-http.client :as http]
   [clojure.string :as str]
   [clojure.walk :as walk]
   [integrant.core :as ig]
   [ring.util.codec :as codec]
   [ring.util.response :as response]))

(def tokens-file-path "/Users/dliman/tokens.json")
(def schwab-base-url "https://api.schwabapi.com/")
(def schwab-client-id "")
(def schwab-client-secret "")
(def schwab-callback-url "https://c02589953629.ngrok-free.app/schwab/oauth")

(def conn (atom nil))
(def tokens (atom nil))

(defn save-tokens [response]
  (swap! tokens (constantly response))
  (try
    (spit tokens-file-path (json/generate-string @tokens {:pretty true}))
    (catch Exception e
      #d [:spit e])))

(defn format-response [response]
  (let [body (:body response)]
    (if (string? body) ;; error
      (walk/keywordize-keys (json/decode body))
      body)))

(defn fetch-access-token [authorization-code]
  (format-response
   (http/post
    (str schwab-base-url "v1/oauth/token")
    {:form-params {:grant_type "authorization_code"
                   :code authorization-code
                   :redirect_uri schwab-callback-url}
     :basic-auth [schwab-client-id schwab-client-secret]
     :content-type :x-www-form-urlencoded
     :throw-exceptions false
     :as :json})))

(defn refresh-access-token [refresh-token]
  (format-response
   (http/post
    (str schwab-base-url "v1/oauth/token")
    {:form-params {:grant_type "refresh_token"
                   :refresh_token refresh-token}
     :basic-auth [schwab-client-id schwab-client-secret]
     :content-type :x-www-form-urlencoded
     :throw-exceptions false
     :as :json})))

(defn user-preference []
  (format-response
   (http/get
    (str schwab-base-url "trader/v1/userPreference")
    {:headers {"Authorization" (str "Bearer " (:access_token @tokens))}
     :throw-exceptions false
     :as :json})))

(defn command-login [streamer-info]
  {:service "LEVELONE_EQUITY" #_"ADMIN"
   :command "LOGIN"
   :requestid (random-uuid)
   :SchwabClientCustomerId (:schwabClientCustomerId streamer-info)
   :SchwabClientCorrelId (:schwabClientCorrelId streamer-info)
   :parameters {:Authorization (:access_token @tokens)
                :SchwabClientChannel (:schwabClientChannel streamer-info)
                :SchwabClientFunctionId (:schwabClientFunctionId streamer-info)}})

(defn connect! []
  (let [preference    (user-preference)
        streamer-info (first (:streamerInfo preference))
        socket @(ws/websocket
                 (:streamerSocketUrl streamer-info)
                 {:on-message (fn [_ws msg last?]
                                #d [:on-message msg last?])
                  :on-close (fn [_ws status reason]
                              #d [:on-message status reason])})]
    (swap! conn (constantly socket))))

(defn send! [command]
  (when-let [conn @conn]
    (ws/send! conn (json/encode command))))

#_(-> user-preference :streamerInfo first command-login send!)
#_(do (connect!)
    (-> (user-preference) :streamerInfo first command-login send!))

(defn get-code [{:strs [code]}]
  code)

(defn parse-authorization-response-handler [req]
  (let [authorization-code
        (or
         (-> req :parameters :query :code)
         (-> req :query-params get-code)
         (-> req :params get-code))

        response #d (fetch-access-token authorization-code)]

    (if (:error tokens)
      #d [:parse-authorization (:error tokens)]
      (save-tokens response))

    (response/response {:status :ok})))

(defn quotes [symbols & {:keys [fields indicative] :or {fields "all" indicative false}}]
  (format-response
   (http/get
    (str schwab-base-url "marketdata/v1/quotes?"
         (codec/form-encode
          {:symbols (str/join "," symbols)
           :fields fields
           :indicative indicative}))
    {:headers {"Authorization" (str "Bearer " (:access_token @tokens))}
     :throw-exceptions false
     :as :json})))

(defmethod ig/init-key ::access-token-refresher [_ _]
  (future
    (loop []
      (when-let [refresh-token (:refresh_token @tokens)]
        (prn "refreshing token...")
        (save-tokens (refresh-access-token refresh-token))
        (prn "refreshed"))
      (Thread/sleep (* 15 60 1000))
      (recur))))

(defmethod ig/halt-key! ::access-token-refresher [_ f]
  (when f
    (future-cancel f)))

(defmethod ig/init-key ::conn [_ _]
  )

(defmethod ig/halt-key! ::conn [_ conn]
  (when conn
    (.close conn)))